package service

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"golang.org/x/sync/errgroup"
	"gorm.io/gorm"
	"os"
	"path"
	"strconv"
	"strings"
	"sync"
	"time"
	"vrcm/src/vrcm/app/admin/models"
	"vrcm/src/vrcm/app/admin/models/live_schedule"
	"vrcm/src/vrcm/app/admin/models/monitor"
	alertService "vrcm/src/vrcm/app/admin/service/alert"
	baiduService "vrcm/src/vrcm/app/admin/service/baidu"
	alertDto "vrcm/src/vrcm/app/admin/service/dto/alert"
	baiduDto "vrcm/src/vrcm/app/admin/service/dto/baidu"
	"vrcm/src/vrcm/app/admin/service/dto/live_schedule"
	vrcDto "vrcm/src/vrcm/app/admin/service/dto/vrc"
	streamHubService "vrcm/src/vrcm/app/admin/service/stream_hub"
	vrcService "vrcm/src/vrcm/app/admin/service/vrc"
	"vrcm/src/vrcm/common/apis"
	"vrcm/src/vrcm/common/service"
	ext "vrcm/src/vrcm/envconfig"

	liveDto "vrcm/src/vrcm/app/admin/service/dto/live"
	StreamHubDto "vrcm/src/vrcm/app/admin/service/dto/stream_hub"
	liveService "vrcm/src/vrcm/app/admin/service/live"
	cDto "vrcm/src/vrcm/common/dto"
)

// 协议类型1rtmp 2hls 3https 4srt 5udp 6sdi
var protocols = map[int]string{
	1: "rtmp",
	2: "hls",
	3: "httpts",
	4: "srt",
	5: "udp",
	6: "flv",
	7: "sdi",
}

// 协议类型1rtmp 2hls 3https 4srt 5udp 6sdi
var streamTypes = map[int]string{
	1: "push",
	2: "pull",
}

var templateMap = map[int]int{
	1: 1,
	2: 2,
	3: 4,
	4: 9,
	5: 16,
	6: 41,
	7: 51,
	8: 71,
}

var errorSourceMap = make(map[int]struct{})

type SourceBasicService struct {
	service.Service
}

// GetPage 获取SourceBasic列表
func (e *SourceBasicService) GetPage(c *dto.SourceBasicGetPageReq, count *int64) (*[]dto.SourceModelRes, error) {
	var err error
	var data live_schedule.SourceBasic
	list := make([]live_schedule.SourceBasic, 0, 5)
	res := make([]dto.SourceModelRes, 0, 5)
	now := time.Now().Unix()
	c.IsDelete = 2
	//根据组查询到的信源
	ids := make([]int, 0, 5)
	groupSourceIds := make([]int, 0, 5)

	if len(c.SourceIds) > 0 {
		//如果是分享信源并且按组查询
		if c.IsShare != 1 && len(groupSourceIds) > 0 {
			for _, v1 := range groupSourceIds {
				for _, v2 := range c.SourceIds {
					if v1 == v2 {
						ids = append(ids, v1)
						break
					}
				}
			}
		} else {
			ids = append(ids, groupSourceIds...)
			ids = append(ids, c.SourceIds...)
		}

	} else {
		ids = append(ids, groupSourceIds...)
	}

	if len(c.GroupIds) > 0 && len(ids) <= 0 {
		return &res, err
	}

	db := e.Orm.Model(&data).
		Scopes(
			cDto.MakeCondition(c.GetNeedSearch()),
			cDto.Paginate(c.GetPageSize(), c.GetPageIndex()),
		).Order("create_time desc")
	if len(ids) > 0 {
		db.Where("id in ?", ids)
	}
	if len(c.TenantId) > 0 && c.IsShare != 2 {
		db.Where("tenant_id = ?", c.TenantId)
	}
	//kind:1 查 非自定义。 kind:2 自定义。 kind:3 查全部
	if c.Kind < 3 {
		if c.Kind > 0 {
			db.Where("kind = ?", c.Kind)
		} else {
			db.Where("kind = 1")
		}
	}

	err = db.Find(&list).Limit(-1).Offset(-1).
		Count(count).Error
	if err != nil {
		e.Log.Errorf("SourceBasicService GetPage error:%s \r\n", err)
		return nil, err
	}

	basicMap := make(map[int]live_schedule.SourceBasic)
	sourceIds := make([]int, 0, 5)
	if len(c.SourceIds) <= 0 {
		for _, v := range list {
			sourceIds = append(sourceIds, v.Id)
			basicMap[v.Id] = v
		}
	} else {
		sourceIds = c.SourceIds
	}
	if len(sourceIds) <= 0 && c.IsShare != 1 {
		return nil, nil
	}
	//获取信源详情
	details := make([]live_schedule.SourceDetail, 0, 5)
	err = e.Orm.Model(&live_schedule.SourceDetail{}).Where("source_id in ? and area !=2", sourceIds).Find(&details).Error
	if err != nil {
		e.Log.Errorf("get SourceDetail error:%s \r\n", err)
		return nil, err
	}

	shareMap := make(map[int]struct{})
	if c.IsShare == 1 {
		shareIds := make([]int, 0, 5)
		err = e.Orm.Model(live_schedule.SourceShare{}).Select("source_id").Where("(account=? or account=?) and share_account !=?", c.TenantId, models.ShareType, c.TenantId).Find(&shareIds).Error
		if err != nil {
			e.Log.Errorf("get SourceShare error:%s \r\n", err)
			return nil, err
		}
		for _, id := range shareIds {
			shareMap[id] = struct{}{}
		}

		realIds := make([]int, 0, 5)
		for _, v := range shareIds {
			isExist := false
			for _, shareSourceId := range sourceIds {
				if v == shareSourceId {
					isExist = true
				}
			}
			if !isExist {
				realIds = append(realIds, v)
			}

		}
		shareSources := make([]live_schedule.SourceBasic, 0, 5)
		err = e.Orm.Model(live_schedule.SourceBasic{}).Where("id in ?", realIds).Find(&shareSources).Error
		if err != nil {
			e.Log.Errorf("get SourceBasic error:%s \r\n", err)
			return nil, err
		}

		list = append(list, shareSources...)
	}

	err = e.Orm.Model(&live_schedule.SourceShare{}).Where("source_id in ?", sourceIds).Error
	if err != nil {
		e.Log.Errorf("get SourceShare error:%s \r\n", err)
		return nil, err
	}
	for _, v := range list {
		v := v
		r := dto.SourceModelRes{}
		r.Basic = &v

		for _, d := range details {

			d := d
			if v.Id != d.SourceId {
				continue
			}
			if v.SourceState != 2 {
				//如果状态是正常 且该状态的更新时间早于当前时间30秒 则状态为正常
				if d.NodeState == 1 && d.StateUpdateTime+10 < now {
					r.Basic.SourceState = 1
				} else {
					r.Basic.SourceState = 3
				}
			}

			detailModel := live_schedule.SourceDetailModel{}
			//如果信源为udp 	且信源启用 默认为正常
			if v.ProtocolType == 5 && v.SourceState != 2 {
				r.Basic.SourceState = 1
			}

			//未启用状态都设置为未配置
			//if d.NodeState == 3 {
			//	detailModel.InSourceState = 1
			//	continue
			//}
			if d.UrlType == 1 {
				detailModel.SourceDetail = d
				r.Details = append(r.Details, detailModel)

				break
			}

		}

		r.IsShare = c.IsShare
		res = append(res, r)

	}
	//*count = int64(len(res))
	return &res, nil
}

// Get 获取SourceBasic对象
func (e *SourceBasicService) Get(d *dto.SourceBasicGetReq) (dto.SourceBasicGetRes, error) {
	var basic live_schedule.SourceBasic
	var res dto.SourceBasicGetRes
	now := time.Now().Unix()
	//获取信源基本信息
	err := e.Orm.Model(&basic).First(&basic, d.GetId()).Error
	if err != nil {
		e.Log.Errorf("Get SourceBasic error:%s", err)
		return res, nil
	}
	res.SourceBasic = &basic

	//获取信源详情
	details := make([]live_schedule.SourceDetail, 0, 4)
	err = e.Orm.Model(&live_schedule.SourceDetail{}).Where("tenant_id=? and source_id=? and area !=2", basic.TenantId, basic.Id).Find(&details).Error
	if err != nil {
		e.Log.Errorf("get SourceDetail error:%s", err)
		res.SourceDetails = nil
	}

	//获取信源状态
	for _, v := range details {
		v := v
		r := live_schedule.SourceDetailModel{}
		r.SourceDetail = v

		if res.SourceBasic.SourceState != 2 {

			//如果状态是正常 且该状态的更新时间早于当前时间10秒 则状态为正常
			if v.NodeState == 1 && v.StateUpdateTime+10 < now {
				v.NodeState = 1
				r.SourceDetail.NodeState = 1
				res.SourceBasic.SourceState = 1
			} else {
				v.NodeState = 2
				r.SourceDetail.NodeState = 2
				res.SourceBasic.SourceState = 3
			}
		}
		res.SourceDetails = append(res.SourceDetails, r)
	}

	//如果信源是udp 默认为正常
	if basic.SourceState != 2 && basic.ProtocolType == 5 {
		basic.SourceState = 1
	}

	return res, nil
}

// GetSourceState 获取信源的状态
func (e *SourceBasicService) GetSourceState(d *dto.SourceBasicGetReq) (int, error) {
	var basic live_schedule.SourceBasic
	now := time.Now().Unix()
	//获取信源基本信息
	err := e.Orm.Model(&basic).First(&basic, d.GetId()).Error
	//获取信源详情
	details := make([]live_schedule.SourceDetail, 0, 4)
	err = e.Orm.Model(&live_schedule.SourceDetail{}).Where("tenant_id=? and source_id=? and area !=2", basic.TenantId, basic.Id).Find(&details).Error
	if err != nil {
		e.Log.Errorf("get SourceDetail error:%s", err)
	}

	//获取信源状态
	for _, v := range details {
		v := v
		//如果状态是正常 且该状态的更新时间早于当前时间10秒 则状态为正常
		if v.NodeState == 1 && v.StateUpdateTime+10 < now {
			v.NodeState = 1
		} else {
			v.NodeState = 2
		}
		if v.IsEnable == 2 && basic.SourceState == 2 {
			v.NodeState = 3
		}
		//主备辅有异常 信源则为异常
		if basic.SourceState == 1 && v.NodeState == 2 && basic.ActiveItem == v.UrlType {
			basic.SourceState = 3
		}
		//如果信源是udp 默认为正常
		if basic.ProtocolType == 5 && v.IsEnable == 1 {
			v.NodeState = 1
		}
		//未启用状态都设置为未配置
		if v.NodeState == 3 {
			continue
		}

		//没有设置未输出节点都为配置未输出
		if v.NodeState == 2 && v.UrlType == basic.ActiveItem {
			// 选中节点为异常 此时信源状态为异常
			if basic.SourceState == 1 {
				basic.SourceState = 3
			}
		}
	}

	////如果信源是udp 默认为正常
	//if basic.SourceState != 2 && basic.ProtocolType == 5 {
	//	basic.SourceState = 1
	//}

	return basic.SourceState, nil
}

// Insert 创建SourceBasic对象
// 推流+外网 先调用全球直播 在调用streamhub 其他情况都直接调用streamhub
func (e *SourceBasicService) Insert(c *dto.SourceBasicInsertReq) error {

	//start := time.Now()
	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	liveService.Log = e.Log
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("SourceBasicService Insert error:%s \r\n", r)
			tx.Rollback()
			//return
		}
	}()
	if err := tx.Error; err != nil {
		e.Log.Errorf("SourceBasicService Insert error:%s \r\n", err)
		return err
	}

	//判断主备辅信源是否存在相同
	m := make(map[string]struct{})
	for _, v := range c.Details {
		if len(v.SourceUrl) <= 0 {
			continue
		}
		_, ok := m[v.SourceUrl]
		if ok {
			return errors.New("流地址不能重复")
		}
		m[v.SourceUrl] = struct{}{}
	}

	//判断信源谁否存在
	var repeatItem live_schedule.SourceBasic
	err := tx.Model(&live_schedule.SourceBasic{}).Where("tenant_id=? and source_name=? and is_delete=2 ", c.TenantId, c.SourceName).Find(&repeatItem).Error

	if err != nil {
		e.Log.Errorf("SourceBasicService Insert error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	if repeatItem.Id > 0 {
		tx.Rollback()
		return errors.New("信源已存在")
	}
	//now := time.Now().Unix()
	var data live_schedule.SourceBasic
	c.Generate(&data)
	err = tx.Create(&data).Error
	if err != nil {
		e.Log.Errorf("SourceBasicService Insert error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	now := time.Now().Unix()
	//组装信源详情
	sourceDetails := GenerateSourceDetailParam(c)

	sources := make([]StreamHubDto.Source, 0, 3)
	//推流+公网 需要调用腾讯直播获取拉流地址
	if c.StreamType == 1 && c.Area == 1 {
		for k, v := range sourceDetails {
			//如果禁用不用调用接口获取udp
			if v.IsEnable == 2 {
				continue
			}

			//否则先去全球直播获取对应的拉流地址
			streamReq := GenerateInsertLiveParam(c)
			res, err := liveService.CreateLiveStream(streamReq)
			if err != nil {
				e.Log.Errorf("CreateLiveStream error:%s \r\n", err)
				tx.Rollback()
				return err
			}
			//更新流地址
			streamType := protocols[c.ProtocolType]
			pushes := strings.Split(res.PushUrl, models.Delimiter)
			*(&sourceDetails[k].PullUrl) = res.RtmpUrl
			*(&sourceDetails[k].LiveId) = res.LiveId
			if len(pushes) > 0 {
				for _, v := range pushes {
					if strings.Contains(v, streamType) {
						*(&sourceDetails[k].PushUrl) = v
						*(&sourceDetails[k].SourceUrl) = res.RtmpUrl
					}
				}
			}
		}
	}

	//组建控制协转接口输入参数
	for _, v := range sourceDetails {
		if v.IsEnable == 2 {
			continue
		}
		source := StreamHubDto.Source{}
		source.Url = v.SourceUrl
		if len(v.PushUrl) > 0 {
			source.Url = v.PullUrl
		}
		sources = append(sources, source)
	}

	//最后调用控制协转接口
	req := StreamHubDto.StreamHubReq{}
	req.Name = c.SourceName
	req.Type = "live"
	req.TenantId = c.TenantId
	req.Description = ""
	req.Input = StreamHubDto.Input{
		Format: protocols[c.ProtocolType],
		Active: 0,
		Policy: StreamHubDto.Policy{
			Mode: "manual",
		},
		Sources: sources,
	}

	//推流+内网 类型为push 否则pull
	if c.StreamType == 1 && c.Area == 2 {
		req.Input.Type = streamTypes[1]
	} else {
		req.Input.Type = streamTypes[2]
	}

	req.CallbackAddress = ext.ExtConfig.ServiceConfig.VrcmServiceUrl + "/api/v1/source-basic/source-call-back"
	delay := StreamHubDto.Delay{
		Enable:   "no",
		Duration: 10,
	}
	req.Output = StreamHubDto.Output{
		Targets: []StreamHubDto.Target{},
		Delay:   delay,
		Acl:     StreamHubDto.Acl{},
	}

	//最后调用hub接口
	hub, err := streamHubService.CreateStreamHub(&req)
	if err != nil {
		e.Log.Errorf("CreateStreamHub error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//更新信源详情
	for k, v := range sourceDetails {
		*(&sourceDetails[k].TaskId) = hub.TaskId
		if c.Area == 3 {
			*(&sourceDetails[k].Area) = 1
		}
		*(&sourceDetails[k].SourceId) = data.Id
		if v.IsEnable == 2 {
			continue
		}
		//记录节点id 整合数据入库
		for _, source := range hub.Input.Sources {
			if source.Url == v.SourceUrl && ((source.Num+1) == sourceDetails[k].UrlType) || len(hub.Input.Sources) == 1 {
				*(&sourceDetails[k].NodeId) = source.Id
				*(&sourceDetails[k].InUdpUrl) = source.UdpAddress
				*(&sourceDetails[k].StateUpdateTime) = now
				*(&sourceDetails[k].SourceUrl) = source.Url
			}
			*(&sourceDetails[k].NodeState) = 3
			//如果是外网推流应该存入推流地址
			if len(v.PushUrl) > 0 {
				*(&sourceDetails[k].SourceUrl) = v.PushUrl
			}

			//单独处理内网推流
			if c.StreamType == 1 && c.Area == 2 && ((source.Num + 1) == sourceDetails[k].UrlType) {
				*(&sourceDetails[k].InUdpUrl) = source.UdpAddress
				*(&sourceDetails[k].NodeId) = source.Id
				*(&sourceDetails[k].SourceUrl) = source.Url
			}

		}
		*(&sourceDetails[k].OutUdpUrl) = hub.OutputAddress
		if len(hub.Output.Targets) > 0 {
			*(&sourceDetails[k].AdditionalPublishUrl) = hub.Output.Targets[0].AdditionalPublishUrl
		}

	}

	if len(sourceDetails) > 0 {
		err = tx.Model(&live_schedule.SourceDetail{}).Create(&sourceDetails).Error
		if err != nil {
			e.Log.Errorf("SourceDetail Insert error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	param := make(map[string]interface{})
	//更新任务id和信源状态
	param["out_udp_url"] = hub.OutputAddress
	param["task_id"] = hub.TaskId
	param["source_state"] = 2
	param["update_time"] = now
	if c.IsCustom == 1 {
		param["out_udp_url"] = c.CustomSwitchUrl
	}
	err = tx.Model(&live_schedule.SourceBasic{}).Where("id=?", data.Id).Updates(param).Error
	if err != nil {
		e.Log.Errorf("SourceBasicService Updates error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//插入分享信息
	conf := live_schedule.ShareSourceConfig{}
	conf.SourceId = data.Id
	conf.TenantId = c.TenantId
	conf.IsEnable = 2
	conf.ShareType = 1
	conf.CreateTime = now
	conf.UpdateTime = now
	conf.CreateBy = c.CreateBy
	err = tx.Model(live_schedule.ShareSourceConfig{}).Create(&conf).Error
	if err != nil {
		e.Log.Errorf("SourceBasicService CreateShareSourceConfig error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	c.Id = data.Id
	tx.Commit()
	return nil
}

func GenerateSourceDetailParam(c *dto.SourceBasicInsertReq) []live_schedule.SourceDetail {

	now := time.Now().Unix()
	sourceDetails := make([]live_schedule.SourceDetail, 0, 5)
	for _, v := range c.Details {
		detail := live_schedule.SourceDetail{}
		//detail.SourceId = data.Id
		detail.UrlType = v.UrlType
		detail.TenantId = c.TenantId
		detail.CreateBy = c.CreateBy
		detail.CreateTime = now
		detail.UpdateTime = now
		detail.CreateBy = c.CreateBy
		detail.IsEnable = v.IsEnable
		detail.SourceUrl = v.SourceUrl
		sourceDetails = append(sourceDetails, detail)
	}
	return sourceDetails
}

func GenerateInsertLiveParam(c *dto.SourceBasicInsertReq) *liveDto.LiveStreamInsertReq {
	streamReq := &liveDto.LiveStreamInsertReq{}
	streamReq.EventName = c.SourceName
	streamReq.TenantId = c.TenantId
	streamReq.ExpireTime = 365 * 24 * 60 * 60
	streamReq.IMEnable = 0
	streamReq.WebPusherEnable = 0
	streamReq.PullOrPush = ""
	streamReq.AutoRecord = "0"
	streamReq.EventType = "1"
	streamReq.ServiceArea = "cn_mainland"
	return streamReq
}

// Update 修改SourceBasic对象
func (e *SourceBasicService) Update(c *dto.SourceBasicUpdateReq) error {
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("SourceBasicService Update error:%s \r\n", r)
			tx.Rollback()
		}
	}()
	if err := tx.Error; err != nil {
		e.Log.Errorf("SourceBasicService Update error:%s \r\n", err)
		return err
	}

	var data = live_schedule.SourceBasic{}
	//更新信息源基本信息
	err := tx.First(&data, c.GetId()).Error
	if err != nil {
		tx.Rollback()
		return err
	}

	if len(data.EquipNo) > 0 {
		err = tx.Model(&live_schedule.SourceBasic{}).Where("id=?", c.Id).Update("source_name", c.SourceName).Error
		if err != nil {
			tx.Rollback()
			return err
		}
	}

	//group := new(errgroup.Group)
	//判断主备辅信源是否存在相同
	m := make(map[string]struct{})
	for _, v := range c.Details {
		if len(v.SourceUrl) <= 0 {
			continue
		}
		_, ok := m[v.SourceUrl]
		if ok {
			return errors.New("流地址不能重复")
		}
		m[v.SourceUrl] = struct{}{}
	}

	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	liveService.Log = e.Log
	channelTaskService := &ChannelTaskService{}
	channelTaskService.Orm = e.Orm
	channelTaskService.Log = e.Log

	//originEnableSwitch := data.EnableAutoSwitch
	c.Generate(&data)
	err = tx.Save(&data).Error
	if err != nil {
		e.Log.Errorf("Update error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//组装信源详情
	sourceDetails, switchDetails, _, isSame, err := e.GenerateUpdateDetailParam(c)
	if err != nil {
		tx.Rollback()
		return err
	}

	//判断是否调用streamhub
	if !isSame {
		//如果是推流且是外网需要调用全球直播接口
		if c.StreamType == 1 && c.Area == 1 {
			for k, v := range sourceDetails {
				//如果是启用且之前没有启用过需要调用全球直播接口
				if v.IsEnable == 1 && len(v.NodeId) <= 0 {

					streamReq := GenerateUpdateLiveParam(c)
					res, err := liveService.CreateLiveStream(streamReq)
					if err != nil {
						e.Log.Errorf("CreateLiveStream error:%s \r\n", err)
						tx.Rollback()
						return err
					}
					streamType := protocols[data.ProtocolType]
					pushes := strings.Split(res.PushUrl, models.Delimiter)
					*(&sourceDetails[k].PullUrl) = res.RtmpUrl
					*(&sourceDetails[k].LiveId) = res.LiveId
					if len(pushes) > 0 {
						for _, v := range pushes {
							if strings.Contains(v, streamType) {
								*(&sourceDetails[k].PushUrl) = v
								*(&sourceDetails[k].SourceUrl) = v
							}
						}
					}
				}
			}
		}

		//组建控制协转接口输入参数
		sources := make([]StreamHubDto.Source, 0, 3)
		for _, v := range sourceDetails {
			if v.IsEnable == 2 {
				continue
			}
			source := StreamHubDto.Source{}
			source.Id = v.NodeId
			source.Url = v.SourceUrl
			if len(v.PushUrl) > 0 {
				source.Url = v.PullUrl
			}
			sources = append(sources, source)
		}

		//最后调用控制协转接口
		req := &StreamHubDto.StreamHubReq{}
		req.Id = data.TaskId
		req.Name = ""
		req.Type = "live"
		req.TenantId = c.TenantId
		req.Description = ""
		req.Input = StreamHubDto.Input{
			Format: protocols[data.ProtocolType],
			Active: c.ActiveItem - 1,
			Policy: StreamHubDto.Policy{
				Mode: "manual",
			},
			Sources: sources,
		}

		delay := StreamHubDto.Delay{
			Enable:   "no",
			Duration: 10,
		}
		req.Output = StreamHubDto.Output{
			Targets: []StreamHubDto.Target{},
			Delay:   delay,
			Acl:     StreamHubDto.Acl{},
		}

		//推流+内网 类型为push 否则pull
		if c.StreamType == 1 && c.Area == 2 {
			req.Input.Type = streamTypes[1]
		} else {
			req.Input.Type = streamTypes[2]
		}

		req.CallbackAddress = ext.ExtConfig.ServiceConfig.VrcmServiceUrl + "/api/v1/source-basic/source-call-back"
		hub, err := streamHubService.UpdateStreamHub(req)
		if err != nil {
			e.Log.Errorf("CreateStreamHub error:%s \r\n", err)
			tx.Rollback()
			return err
		}
		//更新信源详情
		for k, v := range sourceDetails {
			if v.IsEnable == 2 {
				continue
			}
			//记录节点id
			for _, source := range hub.Input.Sources {
				if source.Url == v.SourceUrl || (source.Num+1) == sourceDetails[k].UrlType {
					*(&sourceDetails[k].NodeId) = source.Id
					*(&sourceDetails[k].InUdpUrl) = source.UdpAddress
					*(&sourceDetails[k].SourceUrl) = source.Url
					*(&switchDetails[k].InUdpUrl) = source.UdpAddress
				}
			}

			//如果是外网推流应该存入推流地址
			if len(v.PushUrl) > 0 {
				*(&sourceDetails[k].SourceUrl) = v.PushUrl
			}
			*(&sourceDetails[k].TaskId) = hub.TaskId
			*(&sourceDetails[k].OutUdpUrl) = hub.OutputAddress
			if len(hub.Output.Targets) > 0 {
				*(&sourceDetails[k].AdditionalPublishUrl) = hub.Output.Targets[0].AdditionalPublishUrl
			}
		}
		//更新信源
		if len(sourceDetails) > 0 {
			err = tx.Model(&live_schedule.SourceDetail{}).Save(sourceDetails).Error
			if err != nil {
				e.Log.Errorf("SourceDetail update error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}

	}
	tx.Commit()
	return nil
}

func (e *SourceBasicService) GenerateUpdateDetailParam(c *dto.SourceBasicUpdateReq) ([]live_schedule.SourceDetail, []live_schedule.SourceDetail, []live_schedule.SourceDetail, bool, error) {

	ids := make([]int, 0, 4)
	detailMap := make(map[int]dto.SourceDetailDto)
	details := make([]live_schedule.SourceDetail, 0, 4)
	switchDetails := make([]live_schedule.SourceDetail, 0, 4)

	for _, v := range c.Details {
		v := v
		ids = append(ids, v.Id)
		detailMap[v.Id] = v
	}

	//获取信源的详情
	err := e.Orm.Model(&live_schedule.SourceDetail{}).Where("id in ?", ids).Find(&details).Error
	if err != nil {
		return nil, nil, nil, false, err
	}

	isDiff := false
	for _, v := range details {
		if isDiff == true {
			break
		}
		for _, d := range c.Details {
			if v.UrlType != d.UrlType {
				continue
			}
			if v.IsEnable != d.IsEnable || v.SourceUrl != d.SourceUrl {
				e.Log.Trace("存在不相同的数据")
				isDiff = true
				break
			}
		}
	}

	//需要删除的流
	delDetails := make([]live_schedule.SourceDetail, 0, 5)
	for _, v := range details {
		v := v
		flag := false
		for _, d := range c.Details {
			if v.UrlType == d.UrlType && v.IsEnable == 1 && d.IsEnable == 2 {
				flag = true
				break
			}
		}
		if flag {
			delDetails = append(delDetails, v)
		}
	}

	if !isDiff {
		return nil, nil, nil, true, nil
	}

	//组装详情数据
	for k, v := range details {
		v := v
		detail := detailMap[v.Id]
		//组装编辑切换信源任务参数
		switchDetail := live_schedule.SourceDetail{}
		switchDetail = v

		switchDetail.UrlType = detail.UrlType
		switchDetail.IsEnable = detail.IsEnable
		switchDetail.PushUrl = detail.PushUrl
		//新增
		if len(v.NodeId) <= 0 {
			switchDetail.NodeId = ""
			switchDetail.SwitchNodeId = ""
		} else {
			//修改
			if detail.SourceUrl != v.SourceUrl {
				switchDetail.NodeId = ""
				switchDetail.SwitchNodeId = ""
			}
		}
		//删除
		if detail.IsEnable == 2 {
			switchDetail.NodeId = ""
			switchDetail.SwitchNodeId = ""
		}
		switchDetails = append(switchDetails, switchDetail)

		//组装信源编辑参数
		*(&details[k].IsEnable) = detail.IsEnable
		*(&details[k].SourceUrl) = detail.SourceUrl
		*(&details[k].UpdateTime) = time.Now().Unix()
		*(&details[k].UpdateBy) = c.UpdateBy
		if detail.IsEnable == 2 && len(detail.SourceUrl) <= 0 {
			*(&details[k].NodeId) = ""
			*(&details[k].LiveId) = ""
			*(&details[k].TaskId) = ""
			*(&details[k].OutUdpUrl) = ""
			*(&details[k].InUdpUrl) = ""
			*(&details[k].PushUrl) = ""
			*(&details[k].PullUrl) = ""
			*(&details[k].SwitchNodeId) = ""
		}
	}
	return details, switchDetails, delDetails, false, nil
}

func GenerateUpdateLiveParam(c *dto.SourceBasicUpdateReq) *liveDto.LiveStreamInsertReq {
	streamReq := &liveDto.LiveStreamInsertReq{}
	streamReq.EventName = c.SourceName
	streamReq.TenantId = c.TenantId
	streamReq.ExpireTime = 365 * 24 * 60 * 60
	streamReq.IMEnable = 0
	streamReq.WebPusherEnable = 0
	streamReq.PullOrPush = ""
	streamReq.AutoRecord = "0"
	streamReq.EventType = "1"
	streamReq.ServiceArea = "cn_mainland"
	return streamReq
}

func (e *SourceBasicService) StartSource(d *dto.StartSourceReq) error {

	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("SourceBasicService StartSource error:%s \r\n", r)
			tx.Rollback()
		}
	}()
	if err := tx.Error; err != nil {
		e.Log.Errorf("SourceBasicService StartSource error:%s \r\n", err)
		return err
	}
	baiduService := &baiduService.BaiduService{}
	baiduService.Log = e.Log
	//获取信源基本详情
	//now := time.Now().Unix()
	var basic live_schedule.SourceBasic
	err := tx.Model(live_schedule.SourceBasic{}).Where("id=?", d.Id).Find(&basic).Error
	if err != nil {
		e.Log.Errorf("get source error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	if basic.Id <= 0 {
		tx.Rollback()
		return errors.New("信源不存在或者已被删除")
	}

	//开启信源
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{Id: basic.TaskId, TenantId: d.TenantId})
	if err != nil {
		e.Log.Errorf("StartSource error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//获取信源正在调度中的审核任务
	auditTasks := make([]live_schedule.AuditTask, 0, 5)
	err = tx.Model(&live_schedule.AuditTask{}).Where("source_id=? and audit_task_id!=''", d.Id).Find(&auditTasks).Error
	if err != nil {
		e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	channelIds := make([]int, 0, 5)
	for _, v := range auditTasks {
		v := v
		channelIds = append(channelIds, v.ChannelId)
	}
	//获取正在开启的调度任务
	channelBasics := make([]live_schedule.ChannelBasic, 0, 5)
	err = tx.Model(&live_schedule.ChannelBasic{}).Where("id in ? and channel_state!=2", channelIds).Find(&channelBasics).Error
	if err != nil {
		e.Log.Errorf("get ChannelBasic error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	channelMap := make(map[int]struct{})
	for _, v := range channelBasics {
		v := v
		channelMap[v.Id] = struct{}{}
	}

	auditTaskService := &AuditTaskService{}
	auditTaskService.Orm = e.Orm
	auditTaskService.Log = e.Log
	//获取媒体处理节点
	medias := make([]live_schedule.MediaProcess, 0, 5)
	err = tx.Model(&live_schedule.MediaProcess{}).Where("channel_id in ? and media_type = 3", channelIds).Find(&medias).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get MediaProcess error:%s \r\n", err)
		return err
	}
	mediaMap := make(map[int]live_schedule.MediaProcess)
	for _, v := range medias {
		v := v
		mediaMap[v.Id] = v
	}
	now := time.Now().Unix()
	schedules := make([]live_schedule.ScheduleRecord, 0, 5)
	//审核通过的信源需要重新创建审核任务
	for _, v := range auditTasks {
		_, ok := channelMap[v.ChannelId]
		if v.IsAudit == 1 && ok {
			//重新创建审核任务
			auditId, endTime, err := auditTaskService.CreateAuditTask(v.AuditPullUrl, mediaMap[v.MediaId], baiduService)
			if err != nil {
				tx.Rollback()
				e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
				return err
			}
			param := make(map[string]interface{})
			param["audit_task_id"] = auditId
			param["audit_end_time"] = endTime
			err = tx.Model(&live_schedule.AuditTask{}).Where("id=?", v.Id).Updates(param).Error
			if err != nil {
				tx.Rollback()
				e.Log.Errorf("update AuditTask error:%s \r\n", err)
				return err
			}

			//记录任务启动记录用于统计
			schedule := live_schedule.ScheduleRecord{}
			schedule.TenantId = d.TenantId
			schedule.ChannelId = v.ChannelId
			schedule.ChannelTaskId = v.TaskId
			schedule.StartTime = now
			schedule.Type = 4
			schedule.EndTime = 0
			schedule.CreateTime = now
			schedule.UpdateTime = now
			schedules = append(schedules, schedule)
		}
	}

	//开启原始流切换任务
	if len(basic.OriginSwitchTaskId) > 0 {
		err = streamHubService.StartSwitchSource(&StreamHubDto.StartOrStopSwitchSourceReq{Id: basic.OriginSwitchTaskId, TenantId: d.TenantId})
		if err != nil {
			e.Log.Errorf("StartSwitchSource error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	//更新任务状态
	param := make(map[string]interface{})
	param["source_state"] = 1
	param["start_time"] = time.Now().Unix()
	param["update_time"] = time.Now().Unix()
	err = tx.Model(&live_schedule.SourceBasic{}).Where("id=?", d.Id).Updates(param).Error
	if err != nil {
		e.Log.Errorf("SourceBasicService StartSource error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	if len(schedules) > 0 {
		//保存审核任务开启记录
		err = tx.Model(live_schedule.ScheduleRecord{}).Create(&schedules).Error
		if err != nil {
			e.Log.Errorf("ScheduleRecord insert error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	tx.Commit()
	return nil
}

func (e *SourceBasicService) StopSource(d *dto.StopSourceReq) error {
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("SourceBasicService StopSource error:%s \r\n", r)
			tx.Rollback()
		}
	}()
	if err := tx.Error; err != nil {
		e.Log.Errorf("SourceBasicService StopSource error:%s \r\n", err)
		return err
	}

	var basic live_schedule.SourceBasic
	err := tx.Model(live_schedule.SourceBasic{}).Where("id=?", d.Id).Find(&basic).Error
	if err != nil {
		e.Log.Errorf("get SourceBasic error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//关闭信源
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	baiduService := &baiduService.BaiduService{}
	baiduService.Log = e.Log
	err = streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{Id: basic.TaskId, TenantId: d.TenantId})
	if err != nil {
		e.Log.Errorf("StopSource error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//获取转码列表
	transTemplates := make([]live_schedule.SourceTemplate, 0, 5)
	err = tx.Model(live_schedule.SourceTemplate{}).Select("id").Where("source_id=?", basic.Id).Find(&transTemplates).Error
	if err != nil {
		e.Log.Errorf("get SourceTemplate error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	transTemplateIds := make([]int, 0, 5)
	for _, v := range transTemplates {
		transTemplateIds = append(transTemplateIds, v.Id)
	}

	//获取所有转码任务
	transTasks := make([]live_schedule.TransCodeTask, 0, 5)
	err = tx.Model(live_schedule.TransCodeTask{}).Where("task_record_id in ?", transTemplateIds).Find(&transTasks).Error
	if err != nil {
		e.Log.Errorf("get SourceTemplate error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	now := time.Now().Unix()
	transTaskIds := make([]string, 0, 5)
	//关闭转码任务
	if len(transTasks) > 0 {
		for _, v := range transTasks {
			//停止转码任务
			if len(v.TaskId) > 0 {
				err = streamHubService.StopTransCode(&StreamHubDto.TransCodeReq{Id: v.TaskId, TenantId: d.TenantId})
				if err != nil {
					e.Log.Errorf("StartTransCode error:%s \r\n", err)
					tx.Rollback()
					return err
				}
			}
			transTaskIds = append(transTaskIds, v.TaskId)
		}
		if len(transTemplateIds) > 0 {
			param := make(map[string]interface{})
			param["is_enable"] = 2
			param["update_time"] = time.Now().Unix()

			err = tx.Model(live_schedule.SourceTemplate{}).Where("id in ?", transTemplateIds).Updates(param).Error
			if err != nil {
				e.Log.Errorf("update TransCode error:%s \r\n", err)
				tx.Rollback()
				return err
			}
			for _, v := range transTemplates {
				if len(v.TransCodeSwitchTaskId) > 0 {
					//停止转码流切换
					err = streamHubService.StopSwitchSource(&StreamHubDto.StartOrStopSwitchSourceReq{Id: v.TransCodeSwitchTaskId, TenantId: d.TenantId})
					if err != nil {
						e.Log.Errorf("StopSwitchSource error:%s \r\n", err)
						tx.Rollback()
						return err
					}
				}
			}
		}
		//更新转码任务记录结束时间
		if len(transTaskIds) > 0 {
			param2 := make(map[string]interface{})
			param2["end_time"] = time.Now().Unix()
			err = tx.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id in ? and end_time = 0 and type =2", transTaskIds).Updates(&param2).Error
			if err != nil {
				e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}
	}

	//停止原始流切换任务
	if len(basic.OriginSwitchTaskId) > 0 {
		err = streamHubService.StopSwitchSource(&StreamHubDto.StartOrStopSwitchSourceReq{Id: basic.OriginSwitchTaskId, TenantId: d.TenantId})
		if err != nil {
			e.Log.Errorf("StopSwitchSource error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	//获取此信源的任务调度
	ids := make([]string, 0, 5)
	err = tx.Model(live_schedule.ChannelTask{}).Select("task_id").Where("tenant_id=? and source_id =?", d.TenantId, d.Id).Find(&ids).Error
	if err != nil {
		e.Log.Errorf("SourceBasicService getChannelTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//更新信源调度的结束时间
	param1 := make(map[string]interface{})
	param1["end_time"] = now
	err = tx.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id in ? and end_time = 0 and type=1", ids).Updates(param1).Error
	if err != nil {
		e.Log.Errorf("SourceBasicService updateScheduleRecord error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	//停止录制
	var task live_schedule.SourceRecordTask
	err = tx.Model(live_schedule.SourceRecordTask{}).Where("source_id=? and tenant_id=? and record_task_id =?", basic.Id, basic.TenantId, basic.RecordTaskId).Find(&task).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get SourceRecordTask error:%s \r\n", err)
		return err
	}
	if task.Id > 0 && len(task.RecordTaskId) > 0 {
		err = streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{Id: task.RecordTaskId, TenantId: basic.TenantId})
		if err != nil {
			e.Log.Errorf("StopStreamHub error:%s \r\n", err)
			tx.Rollback()
			return err
		}

		param := make(map[string]interface{})
		param["record_state"] = 2
		param["update_time"] = now
		err = tx.Model(&live_schedule.SourceRecordTask{}).Where("source_id=?", d.Id).Updates(&param).Error
		if err != nil {
			e.Log.Errorf("SourceRecordTask Updates error:%s \r\n", err)
			tx.Rollback()
			return err
		}

		//更新录制任务记录结束时间
		param2 := make(map[string]interface{})
		param2["end_time"] = time.Now().Unix()
		err = tx.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0 and type =3", task.RecordTaskId).Updates(&param2).Error
		if err != nil {
			e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	//获取信源正在调度中的审核任务
	auditTasks := make([]live_schedule.AuditTask, 0, 5)
	err = tx.Model(&live_schedule.AuditTask{}).Where("source_id=? and audit_task_id!=''", d.Id).Find(&auditTasks).Error
	if err != nil {
		e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	//停止审核任务
	auditScheduleIds := make([]string, 0, 5)
	for _, v := range auditTasks {
		err = baiduService.StopAudit(&baiduDto.StopAuditReq{TenantId: v.TenantId, TaskId: v.AuditTaskId})
		if err != nil {
			e.Log.Errorf("StopAudit error:%s \r\n", err)
			tx.Rollback()
			return err
		}
		auditScheduleIds = append(auditScheduleIds, v.TaskId)
	}

	//更新任务id和状态
	param := make(map[string]interface{})
	param["source_state"] = 2
	param["update_time"] = now
	param["stop_time"] = now
	err = tx.Model(&live_schedule.SourceBasic{}).Where("id=?", d.Id).Updates(&param).Error
	if err != nil {
		e.Log.Errorf("SourceBasicService Updates error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	if len(auditScheduleIds) > 0 {
		//更新审核任务记录结束时间
		param2 := make(map[string]interface{})
		param2["end_time"] = now
		err = tx.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id in ? and end_time = 0 and  type =4", auditScheduleIds).Updates(&param2).Error
		if err != nil {
			e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	tx.Commit()
	return nil
}

func (e *SourceBasicService) ChangeSource(d *dto.ChangeSourceReq) error {

	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("SourceBasicService ChangeSource error:%s \r\n", r)
			tx.Rollback()
		}
	}()
	if err := tx.Error; err != nil {
		e.Log.Errorf("SourceBasicService ChangeSource error:%s \r\n", err)
		return err
	}
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	//获取原始信源
	var basic live_schedule.SourceBasic
	err := tx.Model(&live_schedule.SourceBasic{}).Where("id=?", d.Id).Find(&basic).Error
	if err != nil {
		e.Log.Errorf("get SourceBasic error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	if d.TenantId != basic.TenantId {
		tx.Rollback()
		return errors.New(string(apis.ShareSourceChangeError.Code))
	}
	originDetail := make([]live_schedule.SourceDetail, 0, 5)
	err = tx.Model(&live_schedule.SourceDetail{}).Where("source_id=?", d.Id).Find(&originDetail).Error
	if err != nil {
		e.Log.Errorf("get SourceDetail error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//获取转码总任务
	sourceTemplates := make([]live_schedule.SourceTemplate, 0, 5)
	err = tx.Model(live_schedule.SourceTemplate{}).Where("source_id=?", d.Id).Find(&sourceTemplates).Error
	if err != nil {
		e.Log.Errorf("get SourceTemplate error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	sourceTemplateIds := make([]int, 0, 5)
	for _, v := range sourceTemplates {
		sourceTemplateIds = append(sourceTemplateIds, v.Id)
	}
	//转码信源
	transDetails := make([]live_schedule.TransCodeTask, 0, 5)
	err = tx.Model(&live_schedule.TransCodeTask{}).Where("task_record_id in ?", sourceTemplateIds).Find(&transDetails).Error
	if err != nil {
		e.Log.Errorf("get transDetails error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	count := 0
	for _, v := range originDetail {
		if v.IsEnable == 1 {
			count++
		}
	}
	transDetailMap := make(map[int][]live_schedule.TransCodeTask)
	for _, v := range transDetails {
		v := v
		s, ok := transDetailMap[v.TaskRecordId]
		if !ok {
			s = make([]live_schedule.TransCodeTask, 0, 5)
		}
		s = append(s, v)
		transDetailMap[v.TaskRecordId] = s
	}

	//判断转码任务流个数和原始信源个数是否相同
	for _, v := range transDetailMap {
		v := v
		e.Log.Trace("原始流路数：", len(originDetail), "转码流路数", len(v))
		if count != len(v) {
			return errors.New(string(apis.SourceChangeError.Code))
		}
	}

	//获取信源的调度任务
	tasks := make([]live_schedule.ChannelTask, 0, 5)
	err = tx.Model(live_schedule.ChannelTask{}).Where("source_id=? and tenant_id=?", d.Id, d.TenantId).Find(&tasks).Error
	if err != nil {
		e.Log.Errorf("get ChannelTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//获取信源的软矩阵调度任务
	cortexTasks := make([]live_schedule.CortexTask, 0, 5)
	err = tx.Model(live_schedule.CortexTask{}).Where("source_id=? and tenant_id=?", d.Id, d.TenantId).Find(&cortexTasks).Error
	if err != nil {
		e.Log.Errorf("get cortexTasks error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//获取转码的调度任务
	transCodeTasks := make([]live_schedule.ChannelTransCodeTask, 0, 5)
	err = tx.Model(live_schedule.ChannelTransCodeTask{}).Where("source_id=? and tenant_id=?", d.Id, d.TenantId).Find(&transCodeTasks).Error
	if err != nil {
		e.Log.Errorf("get ChannelTransCodeTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	offIds := make([]int, 0, 5)
	upIds := make([]int, 0, 5)
	for _, v := range tasks {
		offIds = append(offIds, v.TrafficOffId)
		upIds = append(upIds, v.TrafficOnId)
	}

	for _, v := range cortexTasks {
		offIds = append(offIds, v.TrafficOffId)
		upIds = append(upIds, v.TrafficOnId)
	}

	//获取下车点
	offs := make([]live_schedule.TrafficOff, 0, 5)
	err = tx.Model(live_schedule.TrafficOff{}).Where("id in ?", offIds).Find(&offs).Error
	if err != nil {
		e.Log.Errorf("get TrafficOff error:%s \r\n", err)
		tx.Rollback()
	}
	offMap := make(map[int]live_schedule.TrafficOff)
	for _, v := range offs {
		v := v
		offMap[v.Id] = v
	}

	//获取上车点
	ons := make([]live_schedule.TrafficOn, 0, 5)
	err = tx.Model(live_schedule.TrafficOn{}).Where("id in ?", upIds).Find(&ons).Error
	if err != nil {
		e.Log.Errorf("get TrafficOn error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	upMap := make(map[int]live_schedule.TrafficOn)
	for _, v := range ons {
		v := v
		upMap[v.Id] = v
	}

	//原始流有缝切换地址
	outUdpUrl := ""
	//转码任务和url的映射
	outUdpUrlMap := make(map[int]string)

	//无缝切换
	if ext.ExtConfig.ServiceConfig.EnableSwitch == 1 && basic.SwitchType == 1 {
		//原始流切换
		if len(basic.OriginSwitchTaskId) > 0 {
			e.Log.Trace("原始刘切换到：", d.ActiveItem)
			err = streamHubService.SwitchSource(&StreamHubDto.ChangeSourceReq{Active: d.ActiveItem - 1, TenantId: d.TenantId, Id: basic.OriginSwitchTaskId})
			if err != nil {
				e.Log.Errorf("ChangeSource error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}
		//转码流切换
		if len(sourceTemplates) > 0 {
			for _, v := range sourceTemplates {
				//转码任务是离线状态不启用切换
				if v.IsEnable == 2 {
					continue
				}
				e.Log.Trace("转码流刘切换到：", d.ActiveItem)
				err = streamHubService.SwitchSource(&StreamHubDto.ChangeSourceReq{Active: d.ActiveItem - 1, TenantId: d.TenantId, Id: v.TransCodeSwitchTaskId})
				if err != nil {
					e.Log.Errorf("ChangeSource error:%s \r\n", err)
					tx.Rollback()
					return err
				}
				outUdpUrlMap[v.Id] = v.TransCodeOutputAddress
			}
		}
	} else {
		//有缝切换
		//切换后的信源udp
		for _, v := range originDetail {
			if v.UrlType == d.ActiveItem {
				outUdpUrl = v.InUdpUrl
				break
			}
		}
		//切换后的转码udp
		for k, v := range transDetailMap {
			k := k
			for _, transDetail := range v {
				if transDetail.UrlType == d.ActiveItem {
					e.Log.Trace("匹配的切换转码url：", k, transDetail.InUdpUrl)
					outUdpUrlMap[k] = transDetail.InUdpUrl
					break
				}
			}

		}
		//协转调度任务切换
		if len(tasks) > 0 {
			for _, task := range tasks {
				//没有协转任务 跳过
				if len(task.TaskId) <= 0 {
					continue
				}
				off := offMap[task.TrafficOffId]
				up := upMap[task.TrafficOnId]
				//组建控制协转接口输入参数
				sources := make([]StreamHubDto.Source, 0, 3)

				source := StreamHubDto.Source{}
				if up.StreamType == 1 {
					source.Url = outUdpUrl
				} else {
					e.Log.Trace("上车点保存的转码任务id：", up.TransRecordId)
					source.Url = outUdpUrlMap[up.TransRecordId]
				}
				source.Id = task.InNodeId
				sources = append(sources, source)

				//最后调用控制协转接口
				req := &StreamHubDto.StreamHubReq{}
				req.Id = task.TaskId
				req.Name = ""
				req.Type = "live"
				req.TenantId = task.TenantId
				req.Description = ""
				req.Input = StreamHubDto.Input{
					Type:   "push",
					Format: "udp",
					Active: d.ActiveItem - 1,
					Policy: StreamHubDto.Policy{
						Mode: "manual",
					},
					Sources: sources,
				}

				delay := StreamHubDto.Delay{
					Enable:   "no",
					Duration: 10,
				}
				target := StreamHubDto.Target{}
				target.Format = protocols[off.TrafficOffType]
				target.Id = task.OutNodeId
				target.Url = task.PushUrl

				//第三方拉流+内网 输出类型为pull 其他情况为push
				if off.StreamType == 2 && off.Area == 2 {
					target.Type = "pull"
				} else {
					target.Type = "push"
				}

				if off.Area == 1 && off.StreamType == 2 {
					target.Format = "rtmp"
				}

				req.Output = StreamHubDto.Output{
					Targets: []StreamHubDto.Target{target},
					Delay:   delay,
					Acl:     StreamHubDto.Acl{},
				}

				req.CallbackAddress = ext.ExtConfig.ServiceConfig.VrcmServiceUrl + "/api/v1/channel-task/cortex-target-call-back"
				_, err = streamHubService.UpdateStreamHub(req)
				if err != nil {
					e.Log.Errorf("UpdateStreamHub error:%s \r\n", err)
					tx.Rollback()
					return err
				}
			}
		}
		//软矩阵调度任务切换
		if len(cortexTasks) > 0 {
			for _, task := range cortexTasks {

				//没有协转任务 跳过
				if len(task.TaskId) <= 0 {
					continue
				}
				off := offMap[task.TrafficOffId]
				up := upMap[task.TrafficOnId]
				//组建控制协转接口输入参数
				sources := make([]StreamHubDto.Source, 0, 3)

				source := StreamHubDto.Source{}
				if up.StreamType == 1 {
					source.Url = outUdpUrl
				} else {
					e.Log.Trace("上车点保存的转码任务id：", up.TransRecordId, outUdpUrlMap[up.TransRecordId])
					source.Url = outUdpUrlMap[up.TransRecordId]
				}
				source.Id = task.InNodeId
				sources = append(sources, source)

				//最后调用控制协转接口
				req := &StreamHubDto.StreamHubReq{}
				req.Id = task.TaskId
				req.Name = ""
				req.Type = "live"
				req.TenantId = task.TenantId
				req.Description = ""
				req.Input = StreamHubDto.Input{
					Type:   "push",
					Format: "udp",
					Active: d.ActiveItem - 1,
					Policy: StreamHubDto.Policy{
						Mode: "manual",
					},
					Sources: sources,
				}

				delay := StreamHubDto.Delay{
					Enable:   "no",
					Duration: 10,
				}
				target := StreamHubDto.Target{}
				target.Format = protocols[off.TrafficOffType]
				target.Id = task.OutNodeId
				target.Url = task.PushUrl

				//第三方拉流+内网 输出类型为pull 其他情况为push
				if off.StreamType == 2 && off.Area == 2 {
					target.Type = "pull"
				} else {
					target.Type = "push"
				}

				if off.Area == 1 && off.StreamType == 2 {
					target.Format = "rtmp"
				}

				req.Output = StreamHubDto.Output{
					Targets: []StreamHubDto.Target{target},
					Delay:   delay,
					Acl:     StreamHubDto.Acl{},
				}

				req.CallbackAddress = ext.ExtConfig.ServiceConfig.VrcmServiceUrl + "/api/v1/channel-task/cortex-target-call-back"
				_, err = streamHubService.UpdateStreamHub(req)
				if err != nil {
					e.Log.Errorf("UpdateStreamHub error:%s \r\n", err)
					tx.Rollback()
					return err
				}
			}
		}
		//转码调度任务切换
		if len(transCodeTasks) > 0 {
			for _, task := range transCodeTasks {
				//没有协转任务 跳过
				if len(task.TaskId) <= 0 {
					continue
				}
				//如果信源是无缝切换，切换信源不需要重新设置转码任务
				if basic.SwitchType == 1 && ext.ExtConfig.ServiceConfig.EnableSwitch == 1 {
					continue
				}
				off := offMap[task.TrafficOffId]
				up := upMap[task.TrafficOnId]
				//组建控制协转接口输入参数
				sources := make([]StreamHubDto.Source, 0, 3)

				source := StreamHubDto.Source{}
				if up.StreamType == 1 {
					task.InUrl = outUdpUrl
					//source.Url = outUdpUrl
				} else {
					task.InUrl = outUdpUrlMap[up.TransRecordId]
					e.Log.Trace("上车点保存的转码任务id：", up.TransRecordId)
					//source.Url = outUdpUrlMap[up.TransRecordId]
				}

				//删掉原来转码任务
				err = streamHubService.DeleteTransCode(&StreamHubDto.TransCodeReq{Id: task.TransTaskId, TenantId: d.TenantId})
				if err != nil {
					e.Log.Errorf("DeleteTransCode error：", err)
					tx.Rollback()
					return err
				}
				//创建新转码任务
				transCodeTaskService := TransCodeTaskService{}
				transCodeTaskService.Log = e.Log
				transCodeTaskService.Orm = e.Orm
				err = transCodeTaskService.CreateSourceTransCodeTask(basic, &task)
				if err != nil {
					e.Log.Errorf("CreateSourceTransCodeTask error：", err)
					tx.Rollback()
					return err
				}
				source.Url = task.InUrl
				source.Id = task.InNodeId
				sources = append(sources, source)

				//更新协转任务
				req := &StreamHubDto.StreamHubReq{}
				req.Id = task.TaskId
				req.Name = ""
				req.Type = "live"
				req.TenantId = task.TenantId
				req.Description = ""
				req.Input = StreamHubDto.Input{
					Type:   "push",
					Format: "udp",
					Active: d.ActiveItem - 1,
					Policy: StreamHubDto.Policy{
						Mode: "manual",
					},
					Sources: sources,
				}

				delay := StreamHubDto.Delay{
					Enable:   "no",
					Duration: 10,
				}
				target := StreamHubDto.Target{}
				target.Format = protocols[off.TrafficOffType]
				target.Id = task.OutNodeId
				target.Url = task.PushUrl

				//第三方拉流+内网 输出类型为pull 其他情况为push
				if off.StreamType == 2 && off.Area == 2 {
					target.Type = "pull"
				} else {
					target.Type = "push"
				}

				if off.Area == 1 && off.StreamType == 2 {
					target.Format = "rtmp"
				}

				req.Output = StreamHubDto.Output{
					Targets: []StreamHubDto.Target{target},
					Delay:   delay,
					Acl:     StreamHubDto.Acl{},
				}

				req.CallbackAddress = ext.ExtConfig.ServiceConfig.VrcmServiceUrl + "/api/v1/channel-task/cortex-target-call-back"
				_, err = streamHubService.UpdateStreamHub(req)
				if err != nil {
					e.Log.Errorf("UpdateStreamHub error:%s \r\n", err)
					tx.Rollback()
					return err
				}
				err = tx.Save(&task).Error
				if err != nil {
					e.Log.Errorf("Update TransCodeTask error:%s \r\n", err)
					tx.Rollback()
					return err
				}
			}
		}
	}
	if len(sourceTemplateIds) > 0 {
		for _, v := range sourceTemplateIds {
			v := v
			param := make(map[string]interface{})
			param["active_item"] = d.ActiveItem
			param["update_time"] = time.Now().Unix()
			if basic.SwitchType == 2 {
				param["trans_code_output_address"] = outUdpUrlMap[v]
			}
			err = tx.Model(&live_schedule.SourceTemplate{}).Where("id = ?", v).Updates(&param).Error
			if err != nil {
				e.Log.Errorf("SourceTemplate Updates error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}

	}

	//更新任务id
	param := make(map[string]interface{})
	param["active_item"] = d.ActiveItem
	if len(outUdpUrl) > 0 {
		param["out_udp_url"] = outUdpUrl
	}
	param["update_time"] = time.Now().Unix()
	err = tx.Model(&live_schedule.SourceBasic{}).Where("id=?", d.Id).Updates(&param).Error
	if err != nil {
		e.Log.Errorf("SourceBasicService Updates error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	var newBasic live_schedule.SourceBasic
	err = tx.Model(&live_schedule.SourceBasic{}).Where("id =?", d.Id).Find(&newBasic).Error
	if err != nil {
		e.Log.Errorf("SourceBasicService get error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	//更新合屏监看
	err = e.GenerateMonitor(tx, newBasic, outUdpUrlMap)
	if err != nil {
		e.Log.Errorf("GenerateMonitors error:%s \r\n", err)
		//return err
	}
	tx.Commit()
	return nil
}

// GenerateMonitor
func (e *SourceBasicService) GenerateMonitor(tx *gorm.DB, sourceBasic live_schedule.SourceBasic, transMap map[int]string) error {
	configs := make([]monitor.MonitorConfig, 0, 5)
	err := tx.Model(&monitor.MonitorConfig{}).Where("source_id=?", sourceBasic.Id).Find(&configs).Error
	if err != nil {
		e.Log.Errorf("SourceBasicService Updates error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	if len(configs) <= 0 {
		return nil
	}
	var vrcService vrcService.VrcService
	vrcService.Log = e.Log
	//存储受影响的监看配置
	basicMonitorIds := make([]int64, 0, 5)
	repeatMap := make(map[int64]struct{})
	for k, v := range configs {
		v := v
		//不是自动切换跳过
		if v.ActiveItem != 5 {
			continue
		}
		if v.StreamType == 1 {
			e.Log.Trace("udp的地址：", sourceBasic.OutUdpUrl)
			*(&configs[k].StreamUrl) = sourceBasic.OutUdpUrl
		} else {
			*(&configs[k].StreamUrl) = transMap[v.TransRecordId]
		}
		_, ok := repeatMap[v.BasicId]
		if !ok {
			basicMonitorIds = append(basicMonitorIds, v.BasicId)
		}

	}
	//保存切换后的udp地址
	err = tx.Save(&configs).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("save MonitorConfig:%s \r\n", err)
		return err
	}

	basics := make([]monitor.MonitorBasic, 0, 5)
	basicMap := make(map[int]monitor.MonitorBasic)
	err = tx.Model(&monitor.MonitorBasic{}).Where("id in ?", basicMonitorIds).Find(&basics).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get MonitorBasic:%s \r\n", err)
		return err
	}
	for _, v := range basics {
		v := v
		basicMap[v.Id] = v
	}

	monitorConfigs := make([]monitor.MonitorConfig, 0, 5)
	monitorConfigMap := make(map[int64][]monitor.MonitorConfig)
	//重新获取视图配置信息
	err = tx.Model(&monitor.MonitorConfig{}).Where("basic_id in ?", basicMonitorIds).Find(&monitorConfigs).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("获取视图配置失败:%s \r\n", err)
		return err
	}
	for _, v := range monitorConfigs {
		v := v
		c, ok := monitorConfigMap[v.BasicId]
		if !ok {
			c = make([]monitor.MonitorConfig, 0, 5)
		}
		c = append(c, v)
		monitorConfigMap[v.BasicId] = c
	}

	g, _ := errgroup.WithContext(context.Background())
	for _, basic := range basics {
		basic := basic
		monitorConfigs := monitorConfigMap[int64(basic.Id)]
		//if basic.MonitorState == 2 {
		//	continue
		//}
		g.Go(func() error {
			inputs := make([]vrcDto.InputSource, 0, 5)
			for _, monitorConfig := range monitorConfigs {
				//创建或者更新房间
				input := vrcDto.InputSource{}
				input.Protocol = "udp"
				input.Type = "push"
				input.Name = monitorConfig.ViewName
				input.Position = monitorConfig.ViewNo
				input.PushAddress = monitorConfig.StreamUrl
				inputs = append(inputs, input)
			}
			_, err := vrcService.SaveRoom(&vrcDto.SaveRoomReq{Title: basic.MonitorName, TenantId: basic.TenantId, Template: templateMap[basic.TemplateType], Poster: basic.VideoPadding, ServiceType: "21", RoomType: "1", EventId: basic.EventId, InputSources: inputs, Id: basic.TaskId})
			if err != nil {
				e.Log.Errorf("保存监看配置失败:%s", err)
				return err
			}

			if basic.MonitorState == 1 {
				err = vrcService.RestartRoom(&vrcDto.StartRoomReq{TenantId: basic.TenantId, EventId: basic.EventId})
				if err != nil {
					e.Log.Errorf("重启监看配置失败:%s", err)
					return err
				}
			}
			return nil
		})
	}
	// 等待所有任务完成
	if err := g.Wait(); err != nil {
		e.Log.Errorf("保存监看配置失败:%s", err)
		return err
	}
	return nil
}

// Remove 删除SourceBasic
func (e *SourceBasicService) Remove(d *dto.SourceBasicDeleteReq) error {

	liveService := &liveService.LiveBasic{}
	hub := &streamHubService.StreamHubService{}
	hub.Log = e.Log
	liveService.Log = e.Log
	baiduService := &baiduService.BaiduService{}
	baiduService.Log = e.Log
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("SourceBasicService Remove error:%s \r\n", r)
			tx.Rollback()
			//return
		}
	}()
	if err := tx.Error; err != nil {
		e.Log.Errorf("SourceBasicService Remove error:%s \r\n", err)
		return err
	}

	//now := time.Now().Unix()
	basics := make([]live_schedule.SourceBasic, 0, 5)
	//获取信源
	err := tx.Where("id in ?", d.Ids).Find(&basics).Error
	if err != nil {
		e.Log.Errorf("get basics error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//获取信源详情
	details := make([]live_schedule.SourceDetail, 0, 5)
	err = tx.Model(live_schedule.SourceDetail{}).Where("source_id in ?", d.Ids).Find(&details).Error
	if err != nil {
		e.Log.Errorf("get SourceDetail error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//信源任务映射
	taskIdMap := make(map[int][]string)
	repeat := make(map[string]struct{})
	for _, v := range details {
		tasks, ok := taskIdMap[v.SourceId]
		if !ok {
			tasks = make([]string, 0, 5)
		}
		_, ok1 := repeat[v.TaskId]
		if !ok1 {
			tasks = append(tasks, v.TaskId)
			repeat[v.TaskId] = struct{}{}
		}
		taskIdMap[v.SourceId] = tasks
		//关闭全球直播流
		if len(v.LiveId) > 0 {
			err := liveService.RemoveLive(&liveDto.StopLiveReq{TenantId: v.TenantId, LiveId: v.LiveId})
			if err != nil {
				e.Log.Errorf("Service RemoveLive error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}
	}

	//删除的信源是启用状态直接返回
	hasEnable := false
	for _, v := range basics {
		if v.SourceState != 2 {
			hasEnable = true
			break
		}
		//关闭信源
		tasks := taskIdMap[v.Id]
		for _, task := range tasks {
			if len(task) > 0 {
				err := hub.StopStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					e.Log.Errorf("Service StopStreamHub error:%s \r\n", err)
					tx.Rollback()
					return err
				}

				err = hub.DeleteStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					e.Log.Errorf("Service DeleteStreamHub error:%s \r\n", err)
					tx.Rollback()
					return err
				}
			}
		}

		//删除原始切换任务
		if len(v.OriginSwitchTaskId) > 0 {
			err := hub.DeleteSwitchSourceTask(&StreamHubDto.SwitchSourceReq{Id: v.OriginSwitchTaskId, TenantId: v.TenantId})
			if err != nil {
				e.Log.Errorf("DeleteSwitchSourceTask error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}

	}
	if hasEnable {
		tx.Rollback()
		return errors.New("信源已启用，请停用后删除")
	}

	//获取转码列表
	sourceTemplates := make([]live_schedule.SourceTemplate, 0, 5)
	err = tx.Model(live_schedule.SourceTemplate{}).Where("source_id in ?", d.Ids).Find(&sourceTemplates).Error
	if err != nil {
		e.Log.Errorf("get SourceTemplate error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	transTemplateIds := make([]int, 0, 5)
	for _, v := range sourceTemplates {
		transTemplateIds = append(transTemplateIds, v.Id)
	}

	//删除转码切换任务
	if len(sourceTemplates) > 0 {
		for _, v := range sourceTemplates {
			if len(v.TransCodeSwitchTaskId) > 0 {
				//停止转码流切换任务
				err = hub.StopSwitchSource(&StreamHubDto.StartOrStopSwitchSourceReq{Id: v.TransCodeSwitchTaskId, TenantId: v.TenantId})
				if err != nil {
					e.Log.Errorf("StartSwitchSource error:%s \r\n", err)
					tx.Rollback()
					return err
				}

				//删除转码流切换任务
				err = hub.DeleteSwitchSourceTask(&StreamHubDto.SwitchSourceReq{Id: v.TransCodeSwitchTaskId, TenantId: v.TenantId})
				if err != nil {
					e.Log.Errorf("DeleteSwitchSourceTask error:%s \r\n", err)
					tx.Rollback()
					return err
				}
			}
		}
	}

	//获取转码任务
	transCodeTasks := make([]live_schedule.TransCodeTask, 0, 5)
	err = tx.Model(live_schedule.TransCodeTask{}).Where("task_record_id in ?", transTemplateIds).Find(&transCodeTasks).Error
	if err != nil {
		e.Log.Errorf("get SourceTemplate error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//删除转码任务
	transTaskIds := make([]string, 0, 5)
	for _, v := range transCodeTasks {
		if len(v.TaskId) > 0 {
			err := hub.StopTransCode(&StreamHubDto.TransCodeReq{TenantId: v.TenantId, Id: v.TaskId})
			if err != nil {
				e.Log.Errorf("Service StopTransCode error:%s \r\n", err)
				tx.Rollback()
				return err
			}

			err = hub.DeleteTransCode(&StreamHubDto.TransCodeReq{TenantId: v.TenantId, Id: v.TaskId})
			if err != nil {
				e.Log.Errorf("Service DeleteTransCode error:%s \r\n", err)
				tx.Rollback()
				return err
			}
			transTaskIds = append(transTaskIds, v.TaskId)
		}
	}

	//更新转码任务记录结束时间
	if len(transTaskIds) > 0 {
		param2 := make(map[string]interface{})
		param2["end_time"] = time.Now().Unix()
		err = tx.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id in ? and end_time = 0 and type =2", transTaskIds).Updates(&param2).Error
		if err != nil {
			e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	err = tx.Model(&live_schedule.SourceTemplate{}).Where("id in ?", transTemplateIds).Delete(&live_schedule.SourceTemplate{}).Error
	if err != nil {
		e.Log.Errorf("delete SourceTemplate error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	//获取调度任务
	channelTasks := make([]live_schedule.ChannelTask, 0, 5)
	err = tx.Model(&live_schedule.ChannelTask{}).Where("source_id in ?", d.Ids).Find(&channelTasks).Error
	if err != nil {
		e.Log.Errorf("get ChannelTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	for _, v := range channelTasks {
		if len(v.TaskId) > 0 {
			err := hub.StopStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
			if err != nil {
				e.Log.Errorf("Service StopStreamHub error:%s \r\n", err)
				tx.Rollback()
				return err
			}

			err = hub.DeleteStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
			if err != nil {
				e.Log.Errorf("Service DeleteStreamHub error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}

	}

	//删除调度任务
	err = tx.Model(&live_schedule.ChannelTask{}).Where("source_id in ?", d.Ids).Delete(&live_schedule.ChannelTask{}).Error
	if err != nil {
		e.Log.Errorf("delete ChannelTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//获取矩阵任务
	cortexTasks := make([]live_schedule.CortexTask, 0, 5)
	err = tx.Model(&live_schedule.CortexTask{}).Where("source_id in ?", d.Ids).Find(&cortexTasks).Error
	if err != nil {
		e.Log.Errorf("get CortexTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	for _, v := range cortexTasks {
		if len(v.TaskId) > 0 {
			err := hub.StopStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
			if err != nil {
				e.Log.Errorf("Service StopStreamHub error:%s \r\n", err)
				tx.Rollback()
				return err
			}

			err = hub.DeleteStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
			if err != nil {
				e.Log.Errorf("Service DeleteStreamHub error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}

	}
	//删除矩阵任务
	err = tx.Model(&live_schedule.CortexTask{}).Where("source_id in ?", d.Ids).Delete(&live_schedule.CortexTask{}).Error
	if err != nil {
		e.Log.Errorf("delete CortexTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//删除录制任务
	recordTasks := make([]live_schedule.SourceRecordTask, 0, 5)
	err = tx.Model(live_schedule.SourceRecordTask{}).Where("source_id in ?", d.Ids).Find(&recordTasks).Error
	if err != nil {
		e.Log.Errorf("get SourceRecordTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	for _, v := range recordTasks {
		if len(v.RecordTaskId) > 0 {
			err = hub.DeleteStreamHub(&StreamHubDto.StreamHubReq{Id: v.RecordTaskId, TenantId: v.TenantId})
			if err != nil {
				e.Log.Errorf("StopStreamHub error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}

		if len(v.LiveId) > 0 {
			err = liveService.RemoveLive(&liveDto.StopLiveReq{TenantId: v.TenantId, LiveId: v.LiveId})
			if err != nil {
				e.Log.Errorf("Service RemoveLive error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}

		if len(v.SwitchTaskId) > 0 {
			err = hub.DeleteSwitchSourceTask(&StreamHubDto.SwitchSourceReq{Id: v.SwitchTaskId, TenantId: v.TenantId})
			if err != nil {
				e.Log.Errorf("DeleteSwitchSourceTask error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}
	}

	err = tx.Model(live_schedule.SourceRecordTask{}).Where("source_id in ?", d.Ids).Delete(&live_schedule.SourceRecordTask{}).Error
	if err != nil {
		e.Log.Errorf("Service delete SourceRecordTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//获取审核任务
	auditTask := make([]live_schedule.AuditTask, 0, 5)
	err = tx.Model(&live_schedule.AuditTask{}).Where("source_id in ?", d.Ids).Find(&auditTask).Error
	if err != nil {
		e.Log.Errorf("get AuditTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	for _, v := range auditTask {
		if len(v.TaskId) > 0 {
			err = hub.StopStreamHub(&StreamHubDto.StreamHubReq{Id: v.TaskId, TenantId: v.TenantId})
			if err != nil {
				e.Log.Errorf("StopStreamHub error:%s \r\n", err)
				tx.Rollback()
				return err
			}
			err = hub.DeleteStreamHub(&StreamHubDto.StreamHubReq{Id: v.TaskId, TenantId: v.TenantId})
			if err != nil {
				e.Log.Errorf("DeleteStreamHub error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}

		if len(v.LiveId) > 0 {
			err = liveService.RemoveLive(&liveDto.StopLiveReq{TenantId: v.TenantId, LiveId: v.LiveId})
			if err != nil {
				e.Log.Errorf("Service RemoveLive error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}

		if len(v.PullTaskId) > 0 {
			err = hub.StopStreamHub(&StreamHubDto.StreamHubReq{Id: v.TaskId, TenantId: v.TenantId})
			if err != nil {
				e.Log.Errorf("StopStreamHub error:%s \r\n", err)
				tx.Rollback()
				return err
			}
			err = hub.DeleteStreamHub(&StreamHubDto.StreamHubReq{Id: v.PullTaskId, TenantId: v.TenantId})
			if err != nil {
				e.Log.Errorf("DeleteStreamHub error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}
		if len(v.AuditTaskId) > 0 {
			err = baiduService.StopAudit(&baiduDto.StopAuditReq{TenantId: v.TenantId, TaskId: v.AuditTaskId})
			if err != nil {
				e.Log.Errorf("StopAudit error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}
	}
	//删除审核任务
	err = tx.Model(live_schedule.AuditTask{}).Where("source_id in ?", d.Ids).Delete(&live_schedule.AuditTask{}).Error
	if err != nil {
		e.Log.Errorf("delete AuditTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	//获取监看
	var vrcService vrcService.VrcService
	vrcService.Log = e.Log
	//monitorConfigs := make([]monitor.MonitorConfig, 0, 5)
	//monitorBasics := make([]monitor.MonitorBasic, 0, 5)
	//eventMap := make(map[int64][]monitor.MonitorConfig)
	basicIds := make([]int64, 0, 5)
	err = tx.Model(&monitor.MonitorConfig{}).Distinct("basic_id").Select("basic_id").Where("source_id in ?", d.Ids).Find(&basicIds).Error
	if err != nil {
		e.Log.Errorf("get MonitorConfig error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	params := make(map[string]interface{})
	params["stream_url"] = ""
	params["trans_record_id"] = 0
	//params["is_config"] = 2
	params["source_id"] = 0
	err = tx.Model(&monitor.MonitorConfig{}).Where("source_id in ?", d.Ids).Updates(&params).Error
	if err != nil {
		e.Log.Errorf("update MonitorConfig error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//err = tx.Model(&monitor.MonitorConfig{}).Where("basic_id in ?", basicIds).Find(&monitorConfigs).Error
	//if err != nil {
	//	e.Log.Errorf("get MonitorConfig error:%s \r\n", err)
	//	tx.Rollback()
	//	return err
	//}
	//for _, v := range monitorConfigs {
	//	v := v
	//	configs, ok := eventMap[v.BasicId]
	//	if !ok {
	//		configs = make([]monitor.MonitorConfig, 0, 5)
	//	}
	//	configs = append(configs, v)
	//	eventMap[v.BasicId] = configs
	//	basicIds = append(basicIds, v.BasicId)
	//}
	//
	//basicMonitorMap := make(map[int]monitor.MonitorBasic)
	//err = tx.Model(&monitor.MonitorBasic{}).Where("id in ?", basicIds).Find(&monitorBasics).Error
	//if err != nil {
	//	e.Log.Errorf("get MonitorBasic error:%s \r\n", err)
	//	tx.Rollback()
	//	return err
	//}
	//
	//for _, v := range monitorBasics {
	//	v := v
	//	basicMonitorMap[v.Id] = v
	//}
	//
	//for k, c := range eventMap {
	//	c := c
	//	b := basicMonitorMap[int(k)]
	//	inputs := make([]vrcDto.InputSource, 0, 5)
	//	for _, v := range c {
	//		input := vrcDto.InputSource{}
	//		input.Protocol = "udp"
	//		input.Type = "push"
	//		input.Name = v.ViewName
	//		input.Position = v.ViewNo
	//		input.PushAddress = v.StreamUrl
	//		inputs = append(inputs, input)
	//	}
	//	go func() {
	//		_, err := vrcService.SaveRoom(&vrcDto.SaveRoomReq{Title: b.MonitorName, TenantId: b.TenantId, Template: templateMap[b.TemplateType], Poster: b.VideoPadding, ServiceType: "21", RoomType: "1", EventId: b.EventId, InputSources: inputs, Id: b.TaskId})
	//		if err != nil {
	//			//tx.Rollback()
	//			e.Log.Errorf("保存监看配置失败:%s", err)
	//			//return err
	//		}
	//		if b.Id > 0 && b.MonitorState == 1 {
	//			err = vrcService.RestartRoom(&vrcDto.StartRoomReq{TenantId: b.TenantId, EventId: b.EventId})
	//			if err != nil {
	//				//tx.Rollback()
	//				e.Log.Errorf("重启监看配置失败:%s", err)
	//				//return err
	//			}
	//		}
	//	}()
	//
	//}

	//删除信源
	err = tx.Model(live_schedule.SourceBasic{}).Where("id in ?", d.Ids).Delete(&live_schedule.SourceBasic{}).Error
	if err != nil {
		e.Log.Errorf("RemoveSourceBasic error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	//删除信源详情
	err = tx.Model(live_schedule.SourceDetail{}).Where("source_id in ?", d.Ids).Delete(&live_schedule.SourceDetail{}).Error
	if err != nil {
		e.Log.Errorf("RemoveSourceDetail error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	//删除信源关联组
	var relation live_schedule.SourceGroupRelation
	err = tx.Model(live_schedule.SourceGroupRelation{}).Where("source_id in ?", d.Ids).Delete(&relation).Error
	if err != nil {
		e.Log.Errorf("Service delete SourceGroupRelation error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//更新上车点为信源删除
	err = tx.Model(live_schedule.TrafficOn{}).Where("source_id in ?", d.Ids).Update("is_delete", 1).Error
	if err != nil {
		e.Log.Errorf("Update TrafficOn error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	tx.Commit()
	return nil
}

func (e *SourceBasicService) CheckDelSource(c *dto.SourceBasicDeleteReq) (int, error) {
	//获取通道任务
	channel1Ids := make([]int, 0, 5)
	err := e.Orm.Model(live_schedule.ChannelTask{}).Distinct("channel_id").Select("channel_id").Where("source_id in ?", c.Ids).Find(&channel1Ids).Error
	if err != nil {
		e.Log.Errorf("get ChannelTask error:%s \r\n", err)
		return apis.DBQueryError.Code, err
	}

	channel2Ids := make([]int, 0, 5)
	err = e.Orm.Model(live_schedule.CortexTask{}).Distinct("channel_id").Select("channel_id").Where("source_id in ?", c.Ids).Find(&channel2Ids).Error
	if err != nil {
		e.Log.Errorf("get CortexTask error:%s \r\n", err)
		return apis.DBQueryError.Code, err
	}

	channelIds := append(channel1Ids, channel2Ids...)
	if len(channelIds) <= 0 {
		return 0, err
	}

	//获取通道基本信息
	basics := make([]live_schedule.ChannelBasic, 0, 5)
	err = e.Orm.Model(live_schedule.ChannelBasic{}).Where("id in ?", channelIds).Find(&basics).Error
	if err != nil {
		e.Log.Errorf("get ChannelBasic error:%s \r\n", err)
		return apis.DBQueryError.Code, err
	}
	isUsing := false
	for _, v := range basics {
		if v.ChannelState != 2 {
			isUsing = true
			break
		}
	}
	if isUsing {
		return apis.SourceIsUsing.Code, err
	} else {
		return apis.SourceIsUsed.Code, err
	}
}

func (e *SourceBasicService) CheckDelTransSource(c *dto.SourceTransDeleteReq) (int, error) {

	//获取转码上车点
	ups := make([]int, 0, 5)
	err := e.Orm.Model(live_schedule.TrafficOn{}).Select("id").Where("source_id = ? and trans_task_id= ?", c.Id, c.TransTaskId).Find(&ups).Error
	if err != nil {
		e.Log.Errorf("get TrafficOn error:%s \r\n", err)
		return apis.DBQueryError.Code, err
	}

	//获取通道任务
	channelIds := make([]int, 0, 5)
	err = e.Orm.Model(live_schedule.ChannelTask{}).Distinct("channel_id").Select("channel_id").Where("traffic_on_id in ?", ups).Find(&channelIds).Error
	if err != nil {
		e.Log.Errorf("get ChannelTask error:%s \r\n", err)
		return apis.DBQueryError.Code, err
	}

	if len(channelIds) <= 0 {
		return 0, err
	}

	//获取通道基本信息
	basics := make([]live_schedule.ChannelBasic, 0, 5)
	err = e.Orm.Model(live_schedule.ChannelBasic{}).Where("id in ?", channelIds).Find(&basics).Error
	if err != nil {
		e.Log.Errorf("get ChannelBasic error:%s \r\n", err)
		return apis.DBQueryError.Code, err
	}
	isUsing := false
	for _, v := range basics {
		if v.ChannelState != 2 {
			isUsing = true
			break
		}
	}
	if isUsing {
		return apis.SourceIsUsing.Code, err
	} else {
		return apis.SourceIsUsed.Code, err
	}
}

func (e *SourceBasicService) SourceCallBack(c *dto.SourceCallBackReq) error {

	//更新通道状态
	param := make(map[string]interface{})
	if c.TransferState == "Capturing" {
		param["node_state"] = 1
	} else {
		param["node_state"] = 2
	}
	param["state_update_time"] = c.TransferStateLastUpdateTime
	err := e.Orm.Model(live_schedule.SourceDetail{}).Where("node_id=?", c.TaskId).Updates(&param).Error
	if err != nil {
		e.Log.Errorf("SourceCallBack error:%s \r\n", err)
		return err
	}

	var detail live_schedule.SourceDetail
	err = e.Orm.Model(live_schedule.SourceDetail{}).Where("node_id=?", c.TaskId).Find(&detail).Error
	if err != nil {
		e.Log.Errorf("get SourceDetail error:%s \r\n", err)
		return err
	}
	var basic live_schedule.SourceBasic
	err = e.Orm.Model(live_schedule.SourceBasic{}).Where("id=?", detail.SourceId).Find(&basic).Error
	if err != nil {
		e.Log.Errorf("get SourceBasic error:%s \r\n", err)
		return err
	}

	if c.TransferState != "Capturing" {
		if len(basic.TaskId) <= 0 {
			return nil
		}
		_, ok := errorSourceMap[basic.Id]
		if ok {
			return nil
		}
		var count int64
		err = e.Orm.Model(live_schedule.SourceDetail{}).Where("source_id=? and node_id!=''", basic.Id).Count(&count).Error
		if err != nil {
			e.Log.Errorf("get SourceDetail error:%s \r\n", err)
			return err
		}
		warnInfo := alertDto.WarnInfo{}
		warnInfo.Type = "software"
		warnInfo.TenantId = basic.TenantId
		warnInfo.Instance = "媒体调度"
		warnInfo.HappenTime = time.Now().Unix()
		warnInfo.Status = "critical"
		warnInfo.Name = basic.TaskId + "_信源信号异常(43003)"
		warnInfo.Description = fmt.Sprintf("信源(%s)异常，请检测信源", basic.SourceName)
		alertService := alertService.AlertService{}
		alertService.Log = e.Log
		err = alertService.AlertInfo(warnInfo)
		if err != nil {
			e.Log.Errorf("AlertInfo error:%s \r\n", err)
			return err
		}
		if basic.SwitchType == 2 && basic.ActiveItem == 1 && detail.UrlType == 1 && count > 1 {
			err = e.ChangeSource(&dto.ChangeSourceReq{TenantId: basic.TenantId, Id: basic.Id, ActiveItem: 2})
			if err != nil {
				e.Log.Errorf("ChangeSource error:%s \r\n", err)
				return err
			}
		}
		errorSourceMap[basic.Id] = struct{}{}
	} else {
		_, ok := errorSourceMap[basic.Id]
		if ok {
			delete(errorSourceMap, basic.Id)
		}
	}

	return nil
}

func (e *SourceBasicService) GetShareSourceConfig(c *dto.GetShareSourceConfig) (*dto.ShareSourceConfigModel, error) {

	model := dto.ShareSourceConfigModel{}
	//获取分享配置
	var config live_schedule.ShareSourceConfig
	err := e.Orm.Model(live_schedule.ShareSourceConfig{}).Where("source_id=?", c.SourceId).Find(&config).Error
	if err != nil {
		e.Log.Errorf("GetShareSourceConfig error:%s \r\n", err)
		return nil, err
	}
	model.ShareSourceConfig = config
	//获取分享的账号
	accounts := make([]string, 0, 5)
	if config.ShareType == 2 && config.IsEnable == 1 {
		err = e.Orm.Model(live_schedule.SourceShare{}).Select("account").Where("source_id=?", config.SourceId).Find(&accounts).Error
		if err != nil {
			e.Log.Errorf("GetShareSource error:%s \r\n", err)
			return nil, err
		}

	}
	model.Accounts = accounts
	return &model, err

}

func (e *SourceBasicService) GetShareSources(c *dto.GetShareSourcesReq, count *int64) (*[]dto.SourceModelRes, error) {
	//获取分享的信源
	sourceShares := make([]live_schedule.SourceShare, 0, 5)
	sourceIds := make([]int, 0, 5)
	err := e.Orm.Model(live_schedule.SourceShare{}).Where("(account=? or account=?) and share_account!=?", c.TenantId, models.ShareType, c.TenantId).Find(&sourceShares).Error
	if err != nil {
		e.Log.Errorf("GetShareSources error:%s \r\n", err)
		return nil, err
	}
	for _, v := range sourceShares {
		sourceIds = append(sourceIds, v.SourceId)
	}
	if len(sourceIds) <= 0 {
		return nil, nil
	}
	data, err := e.GetPage(&dto.SourceBasicGetPageReq{Pagination: c.Pagination, SourceIds: sourceIds, IsDelete: 2, SourceName: c.SourceName, SourceType: c.SourceType, SourceState: c.SourceState, GroupIds: c.GroupIds, IsShare: 2, TenantId: c.TenantId}, count)
	if err != nil {
		e.Log.Errorf("GetShareSources error:%s \r\n", err)
		return nil, err
	}
	if data != nil {
		for k, v := range *data {
			for _, t := range sourceShares {
				t := t
				if t.SourceId != v.Basic.Id {
					continue
				}
				*(&(*data)[k].SourceShare) = t
				break
			}
		}
	}

	return data, nil
}

func (e *SourceBasicService) UpdateShareSourceConfig(c *dto.ShareSourceConfigUpdateReq) error {

	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("UpdateShareSourceConfig error:%s \r\n", r)
			tx.Rollback()
			//return
		}
	}()
	if err := tx.Error; err != nil {
		e.Log.Errorf("UpdateShareSourceConfig error:%s \r\n", err)
		return err
	}

	now := time.Now().Unix()
	var err error
	var data = live_schedule.ShareSourceConfig{}
	tx.First(&data, c.GetId())
	c.Generate(&data)

	db := tx.Save(&data)
	if err = db.Error; err != nil {
		e.Log.Errorf("ShareSourceConfigService Save error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//先删除后添加分享记录
	err = tx.Where("source_id=?", data.SourceId).Delete(&live_schedule.SourceShare{}).Error
	if err != nil {
		e.Log.Errorf("deleteShareSources error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	if data.IsEnable == 2 {
		tx.Commit()
		return nil
	}

	if data.ShareType == 1 {
		//公开信源
		share := live_schedule.SourceShare{
			SourceId: data.SourceId,
			Account:  models.ShareType,
		}
		share.ShareAccount = c.TenantId
		share.UpdateTime = now
		share.CreateTime = now
		share.CreateBy = c.UpdateBy
		err = tx.Model(live_schedule.SourceShare{}).Create(&share).Error
		if err != nil {
			e.Log.Errorf("addShareSources error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	} else {
		//部分公开信源
		shares := make([]live_schedule.SourceShare, 0, 5)
		if len(c.Accounts) > 0 {
			for _, v := range c.Accounts {
				share := live_schedule.SourceShare{
					SourceId: data.SourceId,
					Account:  v,
				}
				share.ShareAccount = c.TenantId
				share.UpdateTime = now
				share.CreateTime = now
				share.CreateBy = c.UpdateBy
				shares = append(shares, share)
			}
			err = tx.Model(live_schedule.SourceShare{}).Create(&shares).Error
			if err != nil {
				e.Log.Errorf("addShareSources error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}
	}
	tx.Commit()
	return nil
}

func (e *SourceBasicService) UpdateShareRemark(c *dto.ShareSourceRemarkUpdateReq) error {

	//sourceShare := live_schedule.SourceShare{}
	err := e.Orm.Model(&live_schedule.SourceShare{}).Where("id=?", c.Id).Update("remark", c.Remark).Error
	if err != nil {
		e.Log.Errorf("UpdateShareRemark error:%s ", err)
		return err
	}
	return nil
}

func (e *SourceBasicService) GetSourcesStatistics(c *dto.GetSourcesStatistics) (dto.SourcesStatistic, error) {

	var data dto.SourcesStatistic

	//获取基本信息
	basics := make([]live_schedule.SourceBasic, 0, 5)
	db := e.Orm.Model(live_schedule.SourceBasic{})
	if len(c.TenantId) > 0 {
		db.Where("tenant_id = ?", c.TenantId)
	}
	err := db.Where("is_delete = 2 and kind<=1").Find(&basics).Error
	if err != nil {
		e.Log.Errorf("GetSourceBasic error:%s \r\n", err)
		return data, err
	}
	ids := make([]int, 0, 5)
	for _, v := range basics {
		ids = append(ids, v.Id)
	}

	e.Log.Trace("信源统计信息")
	//获取基本详情
	details := make([]live_schedule.SourceDetail, 0, 5)
	err = e.Orm.Model(live_schedule.SourceDetail{}).Where("source_id in ?", ids).Find(&details).Error
	if err != nil {
		e.Log.Errorf("GetSourceDetail error:%s \r\n", err)
		return data, err
	}
	detailMap := make(map[int][]live_schedule.SourceDetail)
	for _, v := range details {
		v := v
		detail, ok := detailMap[v.SourceId]
		if !ok {
			detail = make([]live_schedule.SourceDetail, 0, 5)
		}
		detail = append(detail, v)
		detailMap[v.SourceId] = detail
	}

	normal := 0
	off := 0
	exception := 0
	now := time.Now().Unix()

	for _, v := range basics {
		isException := false
		if v.SourceState == 2 {
			off++
			continue
		}
		detail := detailMap[v.Id]
		for k, d := range detail {
			if d.NodeState == 1 && d.StateUpdateTime+10 < now {
				d.NodeState = 1
			} else {
				d.NodeState = 2
			}
			//当前选中节点为异常则为异常
			if d.NodeState != 1 && v.ActiveItem == d.UrlType && v.ProtocolType != 5 {
				e.Log.Trace("错误的是：", d.NodeState, d.SourceId, k)
				isException = true
				break
			}
		}
		if isException {
			exception++
			continue
		}
		normal++
	}

	data.TotalCount = len(basics)
	data.NormalCount = normal
	data.OffCount = off
	data.FailCount = exception

	return data, nil
}

func (e *SourceBasicService) SourcesTransCodeTaskPage(c *dto.GetSourceTaskPageReq, count *int64) (*[]live_schedule.SourceTemplateModel, error) {
	data := make([]live_schedule.SourceTemplateModel, 0, 5)
	transCodeTasks := make([]live_schedule.TransCodeTask, 0, 5)
	templates := make([]live_schedule.SourceTemplate, 0, 5)
	db := e.Orm.Model(&live_schedule.SourceTemplate{}).
		Scopes(
			cDto.MakeCondition(c.GetNeedSearch()),
			cDto.Paginate(c.GetPageSize(), c.GetPageIndex()),
		)
	db.Where("source_id in ?", c.SourceIds)
	if len(c.TemplateId) > 0 {
		db.Where("template_id = ?", c.TemplateId)
	}

	if len(c.TransTaskId) > 0 {
		db.Where("trans_code_task_id = ?", c.TransTaskId)
	}
	e.Log.Trace("传入的参数为：", *c)
	err := db.
		Find(&templates).Limit(-1).Offset(-1).
		Count(count).Error
	if err != nil {
		e.Log.Errorf("SourcesTransCodeTaskPage GetPage error:%s \r\n", err)
		return nil, err
	}

	err = e.Orm.Model(live_schedule.TransCodeTask{}).Where("source_id in ?", c.SourceIds).Find(&transCodeTasks).Error
	if err != nil {
		e.Log.Errorf("get TransCodeTask error:%s \r\n", err)
		return nil, err
	}

	now := time.Now().Unix()
	for _, v := range templates {
		v := v
		model := live_schedule.SourceTemplateModel{}
		model.TransCodeTasks = make([]live_schedule.TransCodeTaskModel, 0, 5)
		model.SourceTemplate = v
		exceptionMap := make(map[int]bool)
		for _, t := range transCodeTasks {
			t := t
			transCodeTaskModel := live_schedule.TransCodeTaskModel{}
			transCodeTaskModel.TransCodeTask = t

			//如果状态正常 且该状态的更新时间早于当前时间30秒 则状态为正常
			if t.NodeState == 1 && t.StateUpdateTime+10 < now {
				t.NodeState = 1
			} else {
				t.NodeState = 2
			}

			b := exceptionMap[t.TaskRecordId]
			if t.NodeState == 2 && !b {
				exceptionMap[t.TaskRecordId] = true
			}

			//没有设置未输出节点都为配置未输出
			if t.UrlType != v.ActiveItem {
				transCodeTaskModel.InSourceState = 2
			} else if t.NodeState == 1 {
				//配置为输出节点且正常
				transCodeTaskModel.InSourceState = 3
			} else {
				//配置为输出节点且异常
				transCodeTaskModel.InSourceState = 4
				// 选中节点为异常 此时信源状态为异常
				//if basic.SourceState == 1 {
				//	basic.SourceState = 3
				//}
			}

			if v.SourceId == t.SourceId && v.TemplateId == t.TemplateId && v.Id == t.TaskRecordId {
				model.TransCodeTasks = append(model.TransCodeTasks, transCodeTaskModel)
			}

		}

		//异常状态
		b := exceptionMap[v.Id]
		if model.SourceTemplate.IsEnable == 1 && b {
			model.SourceTemplate.IsEnable = 3
		}

		data = append(data, model)
	}
	return &data, nil

}

func (e *SourceBasicService) CreateSourceTransCodeTask(tx *gorm.DB, c *dto.CreateSourceTaskReq) error {
	now := time.Now().Unix()
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	e.Log.Trace("传入的参数为：", *c)
	if tx == nil {
		tx = e.Orm
	}
	//获取信源
	var basic live_schedule.SourceBasic
	err := tx.Model(live_schedule.SourceBasic{}).Where("id=?", c.SourceId).Find(&basic).Error
	if err != nil {
		e.Log.Errorf("getSourceBasic error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	//获取信源基本信息详情
	details := make([]live_schedule.SourceDetail, 0, 5)
	if c.IsCustom == 2 && len(c.SourceDetails) <= 0 && c.Id <= 0 {
		//非自定义没有转码任务时候
		err = tx.Model(live_schedule.SourceDetail{}).Where("tenant_id =? and source_id =? and is_enable=1", c.TenantId, c.SourceId).Find(&details).Error
		if err != nil {
			e.Log.Errorf("getSourceDetail error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	} else if c.IsCustom == 1 {
		types := make([]int, 0, 5)
		for _, v := range c.CustomTransCodeUrls {
			types = append(types, v.UrlType)
		}
		err = tx.Model(live_schedule.SourceDetail{}).Where("tenant_id =? and source_id =? and is_enable=1 and url_type in ?", c.TenantId, c.SourceId, types).Find(&details).Error
		if err != nil {
			e.Log.Errorf("getSourceDetail error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	} else {
		details = c.SourceDetails
	}

	t := live_schedule.SourceTemplate{}
	if c.Id > 0 {
		//更新任务名称
		if len(c.TaskName) > 0 {
			err = tx.Model(live_schedule.TransCodeTask{}).Where("task_record_id=?", c.Id).Update("task_name", c.TaskName).Error
			if err != nil {
				tx.Rollback()
				e.Log.Errorf("UpdateSourceTransCodeTask error:%s \r\n", err)
				return err
			}
		}
		err = tx.Model(&live_schedule.SourceTemplate{}).Where("id=?", c.Id).Find(&t).Error
		if err != nil {
			e.Log.Errorf("getSourceTemplate error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	} else {
		//保存转码与信源的关系
		t.TenantId = c.TenantId
		t.TemplateId = c.Template.TemplateId
		t.SourceId = c.SourceId
		t.CreateTime = now
		t.UpdateTime = now
		t.IsEnable = 2
		t.ActiveItem = 1
		t.EnableAdvance = c.EnableAdvance
		t.ServiceId = c.Advanced.ServiceId
		t.ServiceName = c.Advanced.ServiceName
		t.ServiceProvider = c.Advanced.ServiceProvider
		t.VideoPid = c.Advanced.VideoPid
		t.AudioPid = c.Advanced.AudioPid
		t.PcrPid = c.Advanced.PcrPid
		t.PmtPid = c.Advanced.PmtPid
		t.SchedulerMode = c.SchedulerMode
		t.ServiceLoadId = c.ServiceLoadId
		t.TaskName = c.TaskName
		err = tx.Model(live_schedule.SourceTemplate{}).Create(&t).Error
		if err != nil {
			e.Log.Errorf("insert SourceTemplate error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	params := make(map[string]interface{})
	transTasks := make([]live_schedule.TransCodeTask, 0, 5)
	schedules := make([]live_schedule.ScheduleRecord, 0, 5)
	//创建转码任务
	processTasks := make([]StreamHubDto.TransCodeReq, 0, 5)
	for _, v := range details {
		if len(v.InUdpUrl) <= 0 {
			continue
		}
		input := StreamHubDto.TransferCodeInput{}
		output := StreamHubDto.TransferCodeOutput{}
		input.Sources = make([]StreamHubDto.TransCodeSource, 0, 5)
		source := StreamHubDto.TransCodeSource{
			Url:           v.InUdpUrl,
			Advanced:      c.Advanced,
			SchedulerMode: c.SchedulerMode,
			ServiceLoadId: c.ServiceLoadId,
		}
		for _, data := range c.CustomTransCodeUrls {
			if v.UrlType == data.UrlType {
				if c.IsCustom == 1 {
					source.UdpAddress = data.CustomUrl
					break
				}
			}
		}
		input.Sources = append(input.Sources, source)
		input.Format = "udp"
		input.Type = "push"
		input.Policy = StreamHubDto.Policy{}
		input.Policy.Mode = "manual"
		input.Active = v.UrlType - 1

		output.Template = c.Template
		req := StreamHubDto.TransCodeReq{
			TenantId:        c.TenantId,
			Name:            "",
			Input:           input,
			Output:          output,
			Type:            "live",
			CallbackAddress: "",
			NgvsaServiceUrl: ext.ExtConfig.ServiceConfig.NgvsaServiceUrl,
			VrcServiceUrl:   ext.ExtConfig.ServiceConfig.VrcServiceUrl,
		}
		req.CallbackAddress = ext.ExtConfig.ServiceConfig.VrcmServiceUrl + "/api/v1/source-basic/transcode-call-back"
		processTasks = append(processTasks, req)
		res, err := streamHubService.CreateTransCode(&req)
		if err != nil {
			e.Log.Errorf("CreateTransCode error:%s \r\n", err)
			tx.Rollback()
			return err
		}
		if res != nil {
			//存储转码任务
			for _, s := range res.Input.Sources {
				transTask := live_schedule.TransCodeTask{}
				transTask.SourceUrl = s.Url
				transTask.TemplateId = c.Template.TemplateId
				transTask.SourceId = c.SourceId
				transTask.TenantId = c.TenantId
				transTask.TaskId = res.Id
				transTask.NodeId = s.Id
				transTask.InUdpUrl = s.UdpAddress
				transTask.NodeState = 3
				transTask.OutUdpUrl = s.UdpAddress
				transTask.UrlType = v.UrlType
				transTask.MonitorId = s.MonitorId
				transTask.TaskName = c.TaskName
				transTask.CustomUrl = s.UdpAddress
				transTask.CreateTime = now
				transTask.UpdateTime = now
				transTask.StateUpdateTime = now
				transTask.TaskRecordId = t.Id
				transTask.IsCustom = c.IsCustom
				transTasks = append(transTasks, transTask)
			}
			//如果转码任务是开启 需要开启转码协转任务
			if t.IsEnable == 1 {
				err = streamHubService.StartTransCode(&StreamHubDto.TransCodeReq{Id: res.Id, TenantId: v.TenantId})
				if err != nil {
					e.Log.Errorf("StartTransCode error:%s \r\n", err)
					tx.Rollback()
					return err
				}
			}
			//记录转码开始时间
			schedule := live_schedule.ScheduleRecord{}
			schedule.Type = 2
			schedule.TenantId = c.TenantId
			schedule.StartTime = now
			schedule.EndTime = 0
			schedule.CreateTime = now
			schedule.UpdateTime = now
			schedule.ChannelId = c.SourceId
			schedule.ChannelTaskId = res.Id
			schedules = append(schedules, schedule)
		}
	}

	if len(transTasks) > 0 {
		err = tx.Model(live_schedule.TransCodeTask{}).Create(&transTasks).Error
		if err != nil {
			e.Log.Errorf("insert TransCodeTask error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}
	tasks := make([]live_schedule.TransCodeTask, 0, 5)
	err = tx.Model(live_schedule.TransCodeTask{}).Where("task_record_id=?", t.Id).Find(&tasks).Error
	if err != nil {
		e.Log.Errorf("insert TransCodeTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	for _, v := range tasks {
		if v.UrlType == t.ActiveItem {
			params["trans_code_output_address"] = v.OutUdpUrl
		}
	}
	if len(schedules) > 0 {
		err = tx.Model(live_schedule.ScheduleRecord{}).Save(&schedules).Error
		if err != nil {
			e.Log.Errorf("save TransCodeTask error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}
	if ext.ExtConfig.ServiceConfig.EnableSwitch == 1 && basic.SwitchType == 1 {
		e.Log.Trace("进入切换模式")
		switchReq := &StreamHubDto.SwitchSourceReq{}
		switchReq.Name = ""
		switchReq.TenantId = c.TenantId
		switchInput := StreamHubDto.SwitchInput{}
		switchInput.Type = "push"
		switchInput.Format = "udp"
		switchInput.Active = 0
		switchInput.Policy = StreamHubDto.Policy{Mode: "manual"}
		switchInput.Sources = make([]StreamHubDto.Source, 0, 3)
		switchReq.Type = "live"
		if c.Id > 0 {
			switchReq.Id = t.TransCodeSwitchTaskId
		}
		for _, v := range tasks {
			if len(v.InUdpUrl) <= 0 || len(v.TaskId) <= 0 {
				continue
			}
			//保存切换任务参数
			switchSource := StreamHubDto.Source{}
			if c.Id > 0 {
				switchSource.Id = v.SwitchNodeId
			}
			switchSource.Url = v.InUdpUrl
			switchInput.Sources = append(switchInput.Sources, switchSource)
		}
		switchReq.SwitchInput = switchInput
		if c.IsCustomSwitch == 1 {
			switchReq.OutputAddress = c.CustomSwitchUrl
		}
		//创建切换任务
		//switchbackRes := &StreamHubDto.StreamHubRes{}
		switchRes := &StreamHubDto.SwitchSourceRes{}
		if len(t.TransCodeSwitchTaskId) > 0 {
			switchRes, err = streamHubService.UpdateSwitchSourceTask(switchReq)
			if err != nil {
				e.Log.Errorf("UpdateSwitchSourceTask error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		} else {
			switchRes, err = streamHubService.CreateSwitchSourceTask(switchReq)
			if err != nil {
				e.Log.Errorf("CreateSwitchSourceTask error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}

		params["update_time"] = now
		if switchRes != nil {
			params["trans_code_switch_task_id"] = switchRes.TaskId
			params["trans_code_output_address"] = switchRes.OutputAddress

			e.Log.Trace("匹配到的切换id", switchRes.Input.Sources)
			e.Log.Trace("匹配到的切换id", tasks)
			//保存切换id
			for _, v := range switchRes.Input.Sources {
				v := v
				for k, t := range tasks {
					if v.Url != t.InUdpUrl {
						continue
					}
					e.Log.Trace("匹配到的切换id", v.Id)
					*(&tasks[k].SwitchNodeId) = v.Id
				}
			}
			err = tx.Save(&tasks).Error
			if err != nil {
				e.Log.Errorf("save TransCodeTask error:%s \r\n", err)
				tx.Rollback()
				return err
			}

		}
	}
	//params["active_item"] = basic.ActiveItem
	err = tx.Model(live_schedule.SourceTemplate{}).Where("id = ?", t.Id).Updates(&params).Error
	if err != nil {
		e.Log.Errorf("update SourceTemplate error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	return nil
}

func (e *SourceBasicService) UpdateSourceTransCodeTask(c *dto.UpdateTransCodeTaskReq) error {
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			fmt.Println("err", r)
			e.Log.Errorf("UpdateSourceTransCodeTask error:%s \r\n", r)
			tx.Rollback()
			//return
		}
	}()
	if err := tx.Error; err != nil {
		e.Log.Errorf("UpdateSourceTransCodeTask error:%s \r\n", err)
		return err
	}

	err := tx.Model(live_schedule.TransCodeTask{}).Where("task_record_id=?", c.Id).Update("task_name", c.TaskName).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("UpdateSourceTransCodeTask error:%s ", err)
		return err
	}
	err = tx.Model(live_schedule.SourceTemplate{}).Where("id=?", c.Id).Update("task_name", c.TaskName).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("UpdateSourceTemplate error:%s \r\n", err)
		return err
	}
	tx.Commit()
	return nil
}

func (e *SourceBasicService) StartSourceTransCodeTask(c *dto.StarTransCodTaskReq) error {
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("StartSourceTransCodeTask error:%s \r\n", r)
			tx.Rollback()
		}
	}()
	if err := tx.Error; err != nil {
		e.Log.Errorf("StartSourceTransCodeTask error:%s \r\n", err)
		return err
	}

	//now := time.Now().Unix()
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	var basic live_schedule.SourceBasic
	err := tx.Model(live_schedule.SourceBasic{}).Where("id=?", c.SourceId).Find(&basic).Error
	if err != nil {
		e.Log.Errorf("get SourceBasic error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	if basic.SourceState == 2 {
		tx.Rollback()
		return errors.New("请先启用信源")
	}

	//获取任务下的所有转码任务
	tasks := make([]live_schedule.TransCodeTask, 0, 5)
	err = e.Orm.Model(&live_schedule.TransCodeTask{}).Where("task_record_id=?", c.Id).Find(&tasks).Error
	if err != nil {
		e.Log.Errorf("get TransCodeTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//now := time.Now().Unix()
	//schedules := make([]live_schedule.ScheduleRecord, 0, 5)

	for _, v := range tasks {
		err = streamHubService.StartTransCode(&StreamHubDto.TransCodeReq{TenantId: c.TenantId, Id: v.TaskId})
		if err != nil {
			e.Log.Errorf("StartTransCode error:%s \r\n", err)
			tx.Rollback()
			return err
		}
		////记录转码开始时间
		//schedule := live_schedule.ScheduleRecord{}
		//schedule.Type = 2
		//schedule.TenantId = c.TenantId
		//schedule.StartTime = now
		//schedule.EndTime = 0
		//schedule.CreateTime = now
		//schedule.UpdateTime = now
		//schedule.ChannelId = c.SourceId
		//schedule.ChannelTaskId = c.TaskId
		//schedules = append(schedules, schedule)
	}

	//获取转码总任务
	sourceTemplate := &live_schedule.SourceTemplate{}
	err = e.Orm.Model(&live_schedule.SourceTemplate{}).Where("id =?", c.Id).Find(&sourceTemplate).Error
	if err != nil {
		e.Log.Errorf("get SourceTemplate error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	//启用切换任务
	if len(sourceTemplate.TransCodeSwitchTaskId) > 0 {
		err = streamHubService.StartSwitchSource(&StreamHubDto.StartOrStopSwitchSourceReq{TenantId: c.TenantId, Id: sourceTemplate.TransCodeSwitchTaskId})
		if err != nil {
			e.Log.Errorf("StartSwitchSource error:%s \r\n", err)
			tx.Rollback()
			return err
		}
		//保持原始流和转码流切换索引一致
		if sourceTemplate.ActiveItem != basic.ActiveItem {
			err = streamHubService.SwitchSource(&StreamHubDto.ChangeSourceReq{TenantId: c.TenantId, Id: sourceTemplate.TransCodeSwitchTaskId, Active: basic.ActiveItem - 1})
			if err != nil {
				e.Log.Errorf("SwitchSource error:%s \r\n", err)
				tx.Rollback()
				return err
			}
		}

	}
	param := make(map[string]interface{})
	param["active_item"] = basic.ActiveItem
	param["is_enable"] = 1
	param["update_time"] = time.Now().Unix()

	err = tx.Model(live_schedule.SourceTemplate{}).Where("id=?", c.Id).Updates(param).Error
	if err != nil {
		e.Log.Errorf("update TransCode error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//err = tx.Model(live_schedule.ScheduleRecord{}).Create(&schedules).Error
	//if err != nil {
	//	e.Log.Errorf("Create ScheduleRecord error:%s \r\n", err)
	//	tx.Rollback()
	//	return err
	//}
	tx.Commit()
	return nil
}

func (e *SourceBasicService) StopSourceTransCodeTask(c *dto.StopTransCodTaskReq) error {
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			fmt.Println("err", r)
			e.Log.Errorf("StopSourceTransCodeTask error:%s \r\n", r)
			tx.Rollback()
		}
	}()
	if err := tx.Error; err != nil {
		e.Log.Errorf("StopSourceTransCodeTask error:%s \r\n", err)
		return err
	}

	//now := time.Now().Unix()
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	//获取任务下的所有转码任务
	tasks := make([]live_schedule.TransCodeTask, 0, 5)
	err := e.Orm.Model(&live_schedule.TransCodeTask{}).Where("task_record_id=?", c.Id).Find(&tasks).Error
	if err != nil {
		e.Log.Errorf("get TransCodeTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	//taskIds := make([]string, 0, 5)
	for _, v := range tasks {
		err = streamHubService.StopTransCode(&StreamHubDto.TransCodeReq{TenantId: c.TenantId, Id: v.TaskId})
		if err != nil {
			e.Log.Errorf("StopTransCode error:%s \r\n", err)
			tx.Rollback()
			return err
		}
		//taskIds = append(taskIds, v.TaskId)
	}

	//获取转码总任务
	sourceTemplate := &live_schedule.SourceTemplate{}
	err = e.Orm.Model(&live_schedule.SourceTemplate{}).Where("id =?", c.Id).Find(&sourceTemplate).Error
	if err != nil {
		e.Log.Errorf("get SourceTemplate error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	//停用切换任务
	if len(sourceTemplate.TransCodeSwitchTaskId) > 0 {
		err = streamHubService.StopSwitchSource(&StreamHubDto.StartOrStopSwitchSourceReq{TenantId: c.TenantId, Id: sourceTemplate.TransCodeSwitchTaskId})
		if err != nil {
			e.Log.Errorf("StopSwitchSource error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	param := make(map[string]interface{})
	param["is_enable"] = 2
	param["update_time"] = time.Now().Unix()

	err = tx.Model(live_schedule.SourceTemplate{}).Where("id =?", c.Id).Updates(param).Error
	if err != nil {
		e.Log.Errorf("update TransCode error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	////更新转码任务记录结束时间
	//param1 := make(map[string]interface{})
	//param1["end_time"] = time.Now().Unix()
	//err = tx.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id in  ? and end_time = 0 and type =2", taskIds).Updates(&param1).Error
	//if err != nil {
	//	e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
	//	tx.Rollback()
	//	return err
	//}

	tx.Commit()
	return nil
}

func (e *SourceBasicService) DeleteSourceTransCodeTask(c *dto.DeleteSourceTaskReq) error {
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			fmt.Println("err", r)
			e.Log.Errorf("DeleteSourceTransCodeTask error:%s \r\n", r)
			tx.Rollback()
		}
	}()
	if err := tx.Error; err != nil {
		e.Log.Errorf("DeleteSourceTransCodeTask error:%s \r\n", err)
		return err
	}
	taskIds := make([]string, 0, 5)
	//now := time.Now().Unix()
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	//获取转码任务
	transTasks := make([]string, 0, 5)
	err := tx.Model(live_schedule.TransCodeTask{}).Distinct("task_id").Select("task_id").Where("task_record_id =?", c.Id).Find(&transTasks).Error
	if err != nil {
		e.Log.Errorf("get TransCodeTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//获取转码列表
	sourceTemplate := live_schedule.SourceTemplate{}
	err = tx.Model(live_schedule.SourceTemplate{}).Where("id =?", c.Id).Find(&sourceTemplate).Error
	if err != nil {
		e.Log.Errorf("get SourceTemplate error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	if len(sourceTemplate.TransCodeSwitchTaskId) > 0 {
		err = streamHubService.StopSwitchSource(&StreamHubDto.StartOrStopSwitchSourceReq{TenantId: c.TenantId, Id: sourceTemplate.TransCodeSwitchTaskId})
		if err != nil {
			e.Log.Errorf("StopSwitchSource error:%s \r\n", err)
			tx.Rollback()
			return err
		}
		err = streamHubService.DeleteSwitchSourceTask(&StreamHubDto.SwitchSourceReq{TenantId: c.TenantId, Id: sourceTemplate.TransCodeSwitchTaskId})
		if err != nil {
			e.Log.Errorf("DeleteSwitchSourceTask error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	//获取转码任务的上车点
	upIds := make([]int, 0, 5)
	if sourceTemplate.Id > 0 {
		err = tx.Model(&live_schedule.TrafficOn{}).Select("id").Where("trans_record_id=?", sourceTemplate.Id).Find(&upIds).Error
		if err != nil {
			e.Log.Errorf("get TrafficOn error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	if len(upIds) > 0 {
		//获取受到影响的调度任务
		channel1Ids := make([]int, 0, 5)
		err = tx.Model(&live_schedule.ChannelTask{}).Distinct("channel_id").Select("channel_id").Where("traffic_on_id in ?", upIds).Find(&channel1Ids).Error
		if err != nil {
			e.Log.Errorf("get ChannelTask error:%s \r\n", err)
			tx.Rollback()
			return err
		}

		channel2Ids := make([]int, 0, 5)
		err = tx.Model(&live_schedule.CortexTask{}).Distinct("channel_id").Select("channel_id").Where("traffic_on_id in ?", upIds).Find(&channel2Ids).Error
		if err != nil {
			e.Log.Errorf("get CortexTask error:%s \r\n", err)
			tx.Rollback()
			return err
		}

		channelIds := append(channel1Ids, channel2Ids...)
		channelBasics := make([]live_schedule.ChannelBasic, 0, 5)
		err = tx.Model(&live_schedule.ChannelBasic{}).Where("id in ? and channel_state!=2", channelIds).Find(&channelBasics).Error
		if err != nil {
			e.Log.Errorf("get ChannelBasic error:%s \r\n", err)
			tx.Rollback()
			return err
		}

		if len(channelBasics) > 0 {
			return errors.New("转码任务正在被调度使用，请先停用")
		}
	}

	//删除转码任务
	for i := 0; i < len(transTasks); i++ {
		err = streamHubService.StopTransCode(&StreamHubDto.TransCodeReq{TenantId: c.TenantId, Id: transTasks[i]})
		if err != nil {
			e.Log.Errorf("StopTransCode error:%s \r\n", err)
			tx.Rollback()
			return err
		}
		err = streamHubService.DeleteTransCode(&StreamHubDto.TransCodeReq{TenantId: c.TenantId, Id: transTasks[i]})
		if err != nil {
			e.Log.Errorf("DeleteTransCode error:%s \r\n", err)
			tx.Rollback()
			return err
		}
		taskIds = append(taskIds, transTasks[i])
	}

	//获取监看
	var vrcService vrcService.VrcService
	vrcService.Log = e.Log
	//monitorConfigs := make([]monitor.MonitorConfig, 0, 5)
	//monitorBasics := make([]monitor.MonitorBasic, 0, 5)
	//eventMap := make(map[int64][]monitor.MonitorConfig)
	basicIds := make([]int64, 0, 5)
	err = tx.Model(&monitor.MonitorConfig{}).Distinct("basic_id").Select("basic_id").Where("trans_record_id = ?", c.Id).Find(&basicIds).Error
	if err != nil {
		e.Log.Errorf("get MonitorConfig error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	params := make(map[string]interface{})
	//params["source_id"] = 0
	params["stream_url"] = ""
	params["trans_record_id"] = 0
	//params["is_config"] = 2
	err = tx.Model(&monitor.MonitorConfig{}).Where("trans_record_id = ?", c.Id).Updates(&params).Error
	if err != nil {
		e.Log.Errorf("update MonitorConfig error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//err = tx.Model(&monitor.MonitorConfig{}).Where("basic_id in ?", basicIds).Find(&monitorConfigs).Error
	//if err != nil {
	//	e.Log.Errorf("get MonitorConfig error:%s \r\n", err)
	//	tx.Rollback()
	//	return err
	//}
	//for _, v := range monitorConfigs {
	//	v := v
	//	configs, ok := eventMap[v.BasicId]
	//	if !ok {
	//		configs = make([]monitor.MonitorConfig, 0, 5)
	//	}
	//	configs = append(configs, v)
	//	eventMap[v.BasicId] = configs
	//	basicIds = append(basicIds, v.BasicId)
	//}
	//
	//basicMonitorMap := make(map[int]monitor.MonitorBasic)
	//err = tx.Model(&monitor.MonitorBasic{}).Where("id in ?", basicIds).Find(&monitorBasics).Error
	//if err != nil {
	//	e.Log.Errorf("get MonitorBasic error:%s \r\n", err)
	//	tx.Rollback()
	//	return err
	//}
	//for _, v := range monitorBasics {
	//	v := v
	//	basicMonitorMap[v.Id] = v
	//}
	//for k, c := range eventMap {
	//	c := c
	//	b := basicMonitorMap[int(k)]
	//	inputs := make([]vrcDto.InputSource, 0, 5)
	//	for _, v := range c {
	//		input := vrcDto.InputSource{}
	//		input.Protocol = "udp"
	//		input.Type = "push"
	//		input.Name = v.ViewName
	//		input.Position = v.ViewNo
	//		input.PushAddress = v.StreamUrl
	//		inputs = append(inputs, input)
	//	}
	//	_, err := vrcService.SaveRoom(&vrcDto.SaveRoomReq{Title: b.MonitorName, TenantId: b.TenantId, Template: templateMap[b.TemplateType], Poster: b.VideoPadding, ServiceType: "21", RoomType: "1", EventId: b.EventId, InputSources: inputs, Id: b.TaskId})
	//	if err != nil {
	//		tx.Rollback()
	//		e.Log.Errorf("保存监看配置失败:%s", err)
	//		return err
	//	}
	//	if b.Id > 0 && b.MonitorState == 1 {
	//		err = vrcService.RestartRoom(&vrcDto.StartRoomReq{TenantId: b.TenantId, EventId: b.EventId})
	//		if err != nil {
	//			tx.Rollback()
	//			e.Log.Errorf("重启监看配置失败:%s", err)
	//			return err
	//		}
	//	}
	//
	//}

	//删除转码记录
	err = tx.Model(&live_schedule.TransCodeTask{}).Where("task_record_id = ?", c.Id).Delete(&live_schedule.TransCodeTask{}).Error
	if err != nil {
		e.Log.Errorf("DeleteTransCode error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	//删除转码和模版关系
	err = tx.Model(&live_schedule.SourceTemplate{}).Where("id =?", c.Id).Delete(&live_schedule.SourceTemplate{}).Error
	if err != nil {
		e.Log.Errorf("DeleteSourceTemplate error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//更新上车点为信源转码删除
	err = tx.Model(live_schedule.TrafficOn{}).Where("trans_record_id = ?", c.Id).Update("is_delete", 1).Error
	if err != nil {
		e.Log.Errorf("Update TrafficOn error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//更新转码任务记录结束时间
	param1 := make(map[string]interface{})
	param1["end_time"] = time.Now().Unix()
	err = tx.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id in  ? and end_time = 0 and type =2", taskIds).Updates(&param1).Error
	if err != nil {
		e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	tx.Commit()
	return nil
}

func (e *SourceBasicService) TransCodeCallBack(c *dto.TransCodeCallBackReq) error {

	//更新通道状态
	param := make(map[string]interface{})
	if c.TransferState == "Capturing" {
		param["node_state"] = 1
	} else {
		param["node_state"] = 2
	}
	param["state_update_time"] = c.TransferStateLastUpdateTime
	err := e.Orm.Model(live_schedule.ChannelTransTask{}).Where("task_id=?", c.TaskId).Updates(param).Error
	if err != nil {
		e.Log.Errorf("TransCodeCallBack error:%s \r\n", err)
		return err
	}
	return nil
}

func (e *SourceBasicService) StartSourceRecord(c *dto.StartSourceRecordReq) error {

	now := time.Now().Unix()
	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	liveService.Log = e.Log
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("SourceBasicService Insert error:%s \r\n", r)
			tx.Rollback()
			//return
		}
	}()
	if err := tx.Error; err != nil {
		e.Log.Errorf("SourceBasicService Insert error:%s \r\n", err)
		return err
	}
	//获取信源基本信息
	var data live_schedule.SourceBasic
	err := tx.Model(live_schedule.SourceBasic{}).Where("id=?", c.SourceId).Find(&data).Error
	if err != nil {
		e.Log.Errorf("get SourceBasic error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//获取信源详情
	details := make([]live_schedule.SourceDetail, 0, 5)
	err = tx.Model(live_schedule.SourceDetail{}).Where("source_id=?", c.SourceId).Find(&details).Error
	if err != nil {
		e.Log.Errorf("get SourceDetail error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//获取录制任务
	var task live_schedule.SourceRecordTask
	err = tx.Model(live_schedule.SourceRecordTask{}).Where("source_id=? and tenant_id=? and record_type =1", c.SourceId, c.TenantId).Find(&task).Error
	if err != nil {
		e.Log.Errorf("get SourceRecordTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	if task.RecordState == 1 {
		tx.Rollback()
		return errors.New("录制已开始")
	}
	//全球直播获取对应的拉流地址
	recordTaskId := ""
	record := live_schedule.SourceRecordTask{}
	//如果没有录制任务需要重新创建录制任务
	streamReq := GenerateInsertLiveParam(&dto.SourceBasicInsertReq{TenantId: c.TenantId, SourceName: data.SourceName})
	streamReq.AutoRecord = "1"
	//1.获取全球直播推流地址
	streamReq.RecordFileCallbackUrl = ext.ExtConfig.ServiceConfig.VrcmServiceUrl + "/api/v1/source-basic/record-file-call-back"
	streamReq.Tag = strconv.Itoa(c.SourceId)
	res, err := liveService.CreateLiveStream(streamReq)
	if err != nil {
		e.Log.Errorf("CreateLiveStream error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	if res != nil {
		record.SourceId = data.Id
		record.TenantId = data.TenantId
		record.CreateTime = now
		record.UpdateTime = now
		record.LiveId = res.LiveId
		record.RecordType = 1
		streamType := "rtmp"
		pushes := strings.Split(res.PushUrl, models.Delimiter)
		if len(pushes) > 0 {
			for _, v := range pushes {
				if strings.Contains(v, streamType) {
					record.PushUrl = v
				}
			}
		}
	}

	//3.创建协转任务
	sources := make([]StreamHubDto.Source, 0, 1)
	url := data.OutUdpUrl
	source := StreamHubDto.Source{
		Url: url,
	}
	sources = append(sources, source)
	req := &StreamHubDto.StreamHubReq{}
	req.Name = data.SourceName
	req.Type = "live"
	req.TenantId = c.TenantId
	req.Description = ""
	req.Input = StreamHubDto.Input{
		Format: "udp",
		Active: 0,
		Policy: StreamHubDto.Policy{
			Mode: "manual",
		},
		Type:    "push",
		Sources: sources,
	}

	delay := StreamHubDto.Delay{
		Enable:   "no",
		Duration: 10,
	}
	outs := make([]StreamHubDto.Target, 0, 1)
	out := StreamHubDto.Target{
		Format: "rtmp",
		Url:    record.PushUrl,
		Type:   "push",
	}
	outs = append(outs, out)
	req.Output = StreamHubDto.Output{
		Targets: outs,
		Delay:   delay,
		Acl:     StreamHubDto.Acl{},
	}
	//最后调用hub接口
	hub, err := streamHubService.CreateStreamHub(req)
	if err != nil {
		e.Log.Errorf("CreateStreamHub error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	if hub != nil {
		record.RecordTaskId = hub.TaskId
		recordTaskId = hub.TaskId
	}
	//创建录制任务
	if data.SourceState == 3 {
		record.RecordState = 3
	} else {
		record.RecordState = 1
	}

	err = tx.Model(live_schedule.SourceRecordTask{}).Create(&record).Error
	if err != nil {
		e.Log.Errorf("CreateSourceRecordTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	//}
	//开启录制
	err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: data.TenantId, Id: recordTaskId})
	if err != nil {
		e.Log.Errorf("StartStreamHub error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//记录录制开始时间
	schedule := live_schedule.ScheduleRecord{}
	schedule.Type = 3
	schedule.TenantId = c.TenantId
	schedule.StartTime = now
	schedule.EndTime = 0
	schedule.CreateTime = now
	schedule.UpdateTime = now
	schedule.ChannelId = c.SourceId
	schedule.ChannelTaskId = recordTaskId
	err = tx.Model(live_schedule.ScheduleRecord{}).Create(&schedule).Error
	if err != nil {
		e.Log.Errorf("Create ScheduleRecord error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//更新录制任务id
	param2 := make(map[string]interface{})
	param2["record_task_id"] = recordTaskId
	param2["record_task_id"] = recordTaskId
	param2["update_time"] = now
	err = tx.Model(live_schedule.SourceBasic{}).Where("id=?", c.SourceId).Updates(&param2).Error
	if err != nil {
		e.Log.Errorf("update SourceBasic error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	tx.Commit()

	return nil
}

func (e *SourceBasicService) GenerateStreamHubReq(sources []StreamHubDto.Source, outs []StreamHubDto.Target, tenantId string, name string, inputFormat string, inputType string, delayTime int) *StreamHubDto.StreamHubReq {
	req := &StreamHubDto.StreamHubReq{}
	req.Name = name
	req.Type = "live"
	req.TenantId = tenantId
	req.Description = ""
	req.Input = StreamHubDto.Input{
		Format: inputFormat,
		Active: 0,
		Policy: StreamHubDto.Policy{
			Mode: "manual",
		},
		Type:    inputType,
		Sources: sources,
	}

	delay := StreamHubDto.Delay{
		Enable:   "no",
		Duration: 10,
	}
	if delayTime > 0 {
		delay.Enable = "yes"
		delay.Duration = delayTime
	}
	req.Output = StreamHubDto.Output{
		Targets: outs,
		Delay:   delay,
		Acl:     StreamHubDto.Acl{},
	}
	return req
}

func (e *SourceBasicService) StopSourceRecord(c *dto.StartSourceRecordReq) error {

	now := time.Now().Unix()
	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	liveService.Log = e.Log
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("SourceBasicService Insert error:%s \r\n", r)
			tx.Rollback()
			//return
		}
	}()
	if err := tx.Error; err != nil {
		e.Log.Errorf("SourceBasicService Insert error:%s \r\n", err)
		return err
	}
	//获取信源基本信息
	var data live_schedule.SourceBasic
	err := tx.Model(live_schedule.SourceBasic{}).Where("id=?", c.SourceId).Find(&data).Error
	if err != nil {
		e.Log.Errorf("StartSourceRecord error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//获取录制任务
	var task live_schedule.SourceRecordTask
	err = tx.Model(live_schedule.SourceRecordTask{}).Where("source_id=? and tenant_id=? and record_task_id = ?", c.SourceId, c.TenantId, data.RecordTaskId).Find(&task).Error
	if err != nil {
		e.Log.Errorf("get SourceRecordTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}
	if task.Id > 0 && len(task.RecordTaskId) > 0 {
		//停止录制
		err = streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{TenantId: data.TenantId, Id: task.RecordTaskId})
		if err != nil {
			e.Log.Errorf("StopStreamHub error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	//if len(task.LiveId) > 0 {
	//	err = streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{TenantId: data.TenantId, Id: task.LiveId})
	//	if err != nil {
	//		e.Log.Errorf("StopStreamHub error:%s \r\n", err)
	//		tx.Rollback()
	//		return err
	//	}
	//}

	//更新录制任务记录结束时间
	param := make(map[string]interface{})
	param["end_time"] = time.Now().Unix()
	err = tx.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id =  ? and end_time = 0 and type =3", task.RecordTaskId).Updates(&param).Error
	if err != nil {
		e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//更新录制任务状态
	param1 := make(map[string]interface{})
	if data.SourceState == 3 {
		param1["record_state"] = 3
	} else {
		param1["record_state"] = 2
	}
	param1["update_time"] = now
	err = tx.Model(live_schedule.SourceRecordTask{}).Where("id=?", task.Id).Updates(&param1).Error
	if err != nil {
		e.Log.Errorf("update SourceRecordTask error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	tx.Commit()

	return nil
}

func (e *SourceBasicService) RecordFileCallBack(c *dto.RecordFileCallBack) error {

	startTime, _ := strconv.Atoi(c.StartTime)
	endTime, _ := strconv.Atoi(c.EndTIme)
	duration, _ := strconv.Atoi(c.Duration)
	s := live_schedule.SourceRecord{}
	now := time.Now().Unix()
	s.LiveId = c.LiveId
	s.SourceId = c.Tag
	s.StartTime = int64(startTime)
	s.EndTime = int64(endTime)
	s.FilePath = c.Path
	s.Duration = duration
	s.CreateTime = now
	s.UpdateTime = now
	s.FileId = c.FileId
	s.Type = path.Ext(s.FilePath)
	s.Type = strings.TrimLeft(s.Type, ".")

	//获取信源
	var basic live_schedule.SourceBasic
	err := e.Orm.Model(live_schedule.SourceBasic{}).Where("id=?", s.SourceId).Find(&basic).Error
	if err != nil {
		e.Log.Errorf("get SourceBasic error:%s \r\n", err)
		return err
	}
	s.SourceName = basic.SourceName
	s.TenantId = basic.TenantId
	err = e.Orm.Model(live_schedule.SourceRecord{}).Create(&s).Error
	if err != nil {
		e.Log.Errorf("create SourceRecordTask error:%s \r\n", err)
		return err
	}

	//更新录制完成状态
	param1 := make(map[string]interface{})
	param1["record_state"] = 2
	param1["update_time"] = now
	err = e.Orm.Model(live_schedule.SourceRecordTask{}).Where("live_id=?", c.LiveId).Updates(&param1).Error
	if err != nil {
		e.Log.Errorf("update SourceRecordTask error:%s \r\n", err)
		return err
	}
	return nil
}

func (e *SourceBasicService) SourceIsUsing(basics []live_schedule.SourceBasic) (bool, error) {

	//组装信源id
	sourceIds := make([]int, 0, 5)
	for _, v := range basics {
		sourceIds = append(sourceIds, v.Id)
	}

	//获取通道任务
	channelIds := make([]int, 0, 5)
	err := e.Orm.Model(live_schedule.ChannelTask{}).Distinct("channel_id").Select("channel_id").Where("source_id in ?", sourceIds).Find(&channelIds).Error
	if err != nil {
		e.Log.Errorf("get ChannelTask error:%s \r\n", err)
		return false, err
	}

	//获取通道
	tasks := make([]live_schedule.ChannelBasic, 0, 5)
	err = e.Orm.Model(live_schedule.ChannelBasic{}).Where("channel_state !=2 and id in ?", channelIds).Find(&tasks).Error
	if err != nil {
		e.Log.Errorf("get  ChannelBasic error:%s \r\n", err)
		return false, err
	}

	if len(tasks) > 0 {
		return true, nil
	}
	return false, nil
}

func (e *SourceBasicService) GetSourceRecordPage(c *dto.SourceRecordGetPageReq, count *int64) (*[]live_schedule.SourceRecord, error) {
	var err error
	list := make([]live_schedule.SourceRecord, 0, 5)
	db := e.Orm.Model(&live_schedule.SourceRecord{}).
		Scopes(
			cDto.MakeCondition(c.GetNeedSearch()),
			cDto.Paginate(c.GetPageSize(), c.GetPageIndex()),
		).Order("update_time desc")
	if c.StartTime > 0 {
		db.Where("start_time>?", c.StartTime)
	}
	if c.EndTime > 0 {
		db.Where("start_time<?", c.EndTime)
	}
	err = db.Find(&list).Limit(-1).Offset(-1).
		Count(count).Error
	if err != nil {
		e.Log.Errorf("SourceBasicService GetSourceRecordPage error:%s \r\n", err)
		return nil, err
	}

	return &list, nil
}

func (e *SourceBasicService) GetSourceRecord(c *dto.GetSourceRecordReq) (*[]live_schedule.SourceRecord, error) {
	records := make([]live_schedule.SourceRecord, 0, 5)
	err := e.Orm.Model(live_schedule.SourceRecord{}).Where("source_id =?", c.SourceId).Find(&records).Error
	if err != nil {
		e.Log.Errorf("SourceBasicService GetSourceRecord error:%s \r\n", err)
		return nil, err
	}

	return &records, nil
}

func (e *SourceBasicService) DeleteSourceRecord(c *dto.SourceRecordDeleteReq) error {

	liveBasicService := &liveService.LiveBasic{}
	liveRecordService := &liveService.LiveRecord{}
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	liveBasicService.Log = e.Log
	liveRecordService.Log = e.Log
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("DeleteSourceRecord error:%s \r\n", r)
			tx.Rollback()
			//return
		}
	}()
	if err := tx.Error; err != nil {
		e.Log.Errorf("SourceBasicService Insert error:%s \r\n", err)
		return err
	}
	//获取录制文件
	var data live_schedule.SourceRecord
	err := tx.Model(live_schedule.SourceRecord{}).Where("id=?", c.Id).Find(&data).Error
	if err != nil {
		e.Log.Errorf("get SourceRecord error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	if len(data.LiveId) > 0 {
		err = liveBasicService.RemoveLive(&liveDto.StopLiveReq{TenantId: data.TenantId, LiveId: data.LiveId})
		if err != nil {
			e.Log.Errorf("RemoveLive error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	//删除录制任务
	err = tx.Model(&live_schedule.SourceRecordTask{}).Where("live_id=?", data.LiveId).Delete(&live_schedule.SourceRecordTask{}).Error
	if err != nil {
		e.Log.Errorf("删除录制文件任务记录:%s \r\n", err)
		tx.Rollback()
		return err
	}

	//删除录制文件
	err = liveRecordService.DeleteLiveRecord(&liveDto.DeleteRecordReq{FileId: data.FileId})
	if err != nil {
		e.Log.Errorf("删除录制文件:%s \r\n", err)
		tx.Rollback()
		return err
	}
	//获取信源录制文件
	err = tx.Model(&live_schedule.SourceRecord{}).Where("id=?", c.Id).Delete(&live_schedule.SourceRecord{}).Error
	if err != nil {
		e.Log.Errorf("删除录制文件记录:%s \r\n", err)
		tx.Rollback()
		return err
	}
	tx.Commit()

	return nil
}

func (e *SourceBasicService) GetAuditStatisticBySource(c *dto.GetAuditResultBySourceReq) (*dto.AuditStatisticRes, error) {
	results := make([]dto.AuditStatisticDetailRes, 0, 5)
	db := e.Orm.Model(&live_schedule.AuditResult{})
	db.Where("audit_type =1")
	if c.SourceId > 0 {
		db.Where("source_id=?", c.SourceId)
	}
	if c.ChannelId > 0 {
		db.Where("channel_id=?", c.SourceId)
	}

	err := db.Select("violation_type,task_id, count(*) as type_num").
		Group("violation_type,task_id").Scan(&results).Error
	if err != nil {
		e.Log.Errorf("create AuditResult error:%s \r\n", err)
		return nil, err
	}

	res := dto.AuditStatisticRes{}
	res.SourceId = c.SourceId
	t1 := 0
	t2 := 0
	t3 := 0
	t4 := 0
	t5 := 0
	t6 := 0
	for _, d := range results {
		switch d.ViolationType {
		case 1:
			t1 += d.TypeNum
		case 2:
			t2 += d.TypeNum
		case 3:
			t3 += d.TypeNum
		case 4:
			t4 += d.TypeNum
		case 5:
			t5 += d.TypeNum
		default:
			t6 += d.TypeNum
		}
	}
	res.Type1Num = t1
	res.Type2Num = t2
	res.Type3Num = t3
	res.Type4Num = t4
	res.Type5Num = t5
	res.Type6Num = t6
	return &res, nil
}

func (e *SourceBasicService) GetAuditStatisticByType(c *dto.GetAuditResultDetailBySourceReq, count *int64) (*[]live_schedule.AuditResult, error) {
	results := make([]live_schedule.AuditResult, 0, 5)
	db := e.Orm.Model(&live_schedule.AuditResult{}).
		Scopes(
			cDto.Paginate(c.PageSize, c.PageIndex),
		).Where("source_id=? and audit_type =1", c.SourceId)
	if len(c.ViolationType) > 0 {
		db.Where("violation_type in ?", c.ViolationType)
	}

	err := db.Find(&results).Limit(-1).Offset(-1).
		Count(count).Error
	if err != nil {
		e.Log.Errorf("GetAuditStatisticByType GetPage error:%s \r\n", err)
		return nil, err
	}
	return &results, nil
}

func (e *SourceBasicService) UpdateAuditState(c *dto.UpdateAuditState) error {
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("UpdateAuditState error:%s \r\n", r)
			tx.Rollback()
		}
	}()
	if err := tx.Error; err != nil {
		tx.Rollback()
		e.Log.Errorf("UpdateAuditState error:%s \r\n", err)
		return err
	}
	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	baiduService := &baiduService.BaiduService{}
	streamHubService.Log = e.Log
	liveService.Log = e.Log
	baiduService.Log = e.Log
	auditTaskService := &AuditTaskService{}
	auditTaskService.Orm = e.Orm
	auditTaskService.Log = e.Log

	err := tx.Model(&live_schedule.SourceBasic{}).
		Where("id=?", c.SourceId).
		Update("audit_state", c.AuditState).
		Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("UpdateAuditState error:%s \r\n", err)
		return err
	}
	now := time.Now().Unix()
	history := live_schedule.AuditHistory{}
	history.SourceID = c.SourceId
	history.AuditState = c.AuditState
	history.TenantID = c.TenantId
	history.Account = c.TenantId
	history.AuditTimeStamp = now
	history.CreateTime = now
	history.UpdateTime = now
	err = tx.Model(&live_schedule.AuditHistory{}).Create(&history).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("create AuditHistory error:%s \r\n", err)
		return err
	}

	//获取信源的审核任务
	tasks := make([]live_schedule.AuditTask, 0, 5)
	err = tx.Model(&live_schedule.AuditTask{}).Where("source_id=?", c.SourceId).Find(&tasks).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get AuditTask error:%s \r\n", err)
		return err
	}
	ids := make([]int, 0, 5)
	channelIds := make([]int, 0, 5)
	for _, v := range tasks {
		ids = append(ids, v.Id)
		channelIds = append(channelIds, v.ChannelId)
	}
	//获取通道
	channelBasics := make([]live_schedule.ChannelBasic, 0, 5)
	err = tx.Model(&live_schedule.ChannelBasic{}).Where("id in ?", channelIds).Find(&channelBasics).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get ChannelBasic error:%s \r\n", err)
		return err
	}
	channelStateMap := make(map[int]int)
	for _, v := range channelBasics {
		v := v
		channelStateMap[v.Id] = v.ChannelState
	}

	//获取媒体处理节点
	medias := make([]live_schedule.MediaProcess, 0, 5)
	err = tx.Model(&live_schedule.MediaProcess{}).Where("channel_id in ? and media_type = 3", channelIds).Find(&medias).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get MediaProcess error:%s \r\n", err)
		return err
	}
	mediaMap := make(map[int]live_schedule.MediaProcess)
	for _, v := range medias {
		v := v
		mediaMap[v.Id] = v
	}

	schedules := make([]live_schedule.ScheduleRecord, 0, 5)
	auditScheduleIds := make([]string, 0, 5)
	if c.AuditState == 2 {
		//停止输出
		for _, v := range tasks {
			if len(v.TaskId) > 0 {
				err = streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					tx.Rollback()
					e.Log.Errorf("StopStreamHub error:%s \r\n", err)
					return err
				}
			}
			if len(v.PullTaskId) > 0 {
				err = streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.PullTaskId})
				if err != nil {
					tx.Rollback()
					e.Log.Errorf("StopStreamHub error:%s \r\n", err)
					return err
				}
			}
			if len(v.AuditTaskId) > 0 {
				err = baiduService.StopAudit(&baiduDto.StopAuditReq{TenantId: v.TenantId, TaskId: v.AuditTaskId})
				if err != nil {
					tx.Rollback()
					e.Log.Errorf("StopAuditReq error:%s \r\n", err)
					return err
				}
				auditScheduleIds = append(auditScheduleIds, v.TaskId)
			}
		}
	}
	////审核任务停用
	//if len(auditIds) > 0 {
	//	err = tx.Model(live_schedule.AuditTask{}).Where("id in ?", auditIds).Update("is_enable", 2).Error
	//	if err != nil {
	//		e.Log.Errorf("update AuditTask error:%s \r\n", err)
	//	}
	//}

	isCreateAuditTask := false
	if c.AuditState == 1 {
		//先判断调度通道是否开启 开启输出
		for _, v := range tasks {
			state := channelStateMap[v.ChannelId]
			if state != 2 && len(v.TaskId) > 0 {
				err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					tx.Rollback()
					e.Log.Errorf("StartStreamHub error:%s \r\n", err)
					return err
				}
				if len(v.PullTaskId) > 0 {
					err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.PullTaskId})
					if err != nil {
						tx.Rollback()
						e.Log.Errorf("StartStreamHub error:%s \r\n", err)
						return err
					}
				}

				//重新创建审核任务 并且同一个信源的审核任务只需要创建一次
				if !isCreateAuditTask && len(v.AuditTaskId) > 0 {
					err = baiduService.StopAudit(&baiduDto.StopAuditReq{TenantId: v.TenantId, TaskId: v.AuditTaskId})
					if err != nil {
						tx.Rollback()
						e.Log.Errorf("StopAuditReq error:%s \r\n", err)
						return err
					}
					auditScheduleIds = append(auditScheduleIds, v.TaskId)
					auditId, endTime, err := auditTaskService.CreateAuditTask(v.AuditPullUrl, mediaMap[v.MediaId], baiduService)
					if err != nil {
						tx.Rollback()
						e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
						return err
					}
					param := make(map[string]interface{})
					param["audit_task_id"] = auditId
					param["audit_end_time"] = endTime
					err = tx.Model(&live_schedule.AuditTask{}).Where("id=?", v.Id).Updates(param).Error
					if err != nil {
						tx.Rollback()
						e.Log.Errorf("update AuditTask error:%s \r\n", err)
						return err
					}
					isCreateAuditTask = true
					//记录任务启动记录用于统计
					schedule := live_schedule.ScheduleRecord{}
					schedule.TenantId = c.TenantId
					schedule.ChannelId = v.ChannelId
					schedule.ChannelTaskId = v.TaskId
					schedule.StartTime = now
					schedule.Type = 4
					schedule.EndTime = 0
					schedule.CreateTime = now
					schedule.UpdateTime = now
					schedules = append(schedules, schedule)
				}

			}
		}
	}

	err = tx.Model(&live_schedule.AuditTask{}).Where("source_id =?", c.SourceId).Update("is_audit", c.AuditState).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("update AuditTask error:%s \r\n", err)
		return err
	}

	if len(auditScheduleIds) > 0 {
		//更新审核任务记录结束时间
		param2 := make(map[string]interface{})
		param2["end_time"] = now
		err = tx.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id in ? and end_time = 0 and  type =4", auditScheduleIds).Updates(&param2).Error
		if err != nil {
			e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	if len(schedules) > 0 {
		//保存审核任务开启记录
		err = tx.Model(live_schedule.ScheduleRecord{}).Create(&schedules).Error
		if err != nil {
			e.Log.Errorf("ScheduleRecord insert error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	tx.Commit()
	return nil
}

func (e *SourceBasicService) GetAuditHistoryBySource(c *dto.GetAuditState) (*[]live_schedule.AuditHistory, error) {

	histories := make([]live_schedule.AuditHistory, 0, 5)
	err := e.Orm.Model(&live_schedule.AuditHistory{}).
		Where("source_id=?", c.SourceId).
		Order("create_time desc").
		Find(&histories).
		Error
	if err != nil {
		e.Log.Errorf("get AuditHistory error:%s \r\n", err)
		return nil, err
	}
	return &histories, nil
}

func (e *SourceBasicService) GetCustomSource(ids []int) (map[int]live_schedule.SourceBasic, error) {
	sourceMap := make(map[int]live_schedule.SourceBasic)
	sources := make([]live_schedule.SourceBasic, 0, 5)
	err := e.Orm.Model(&live_schedule.SourceBasic{}).Where("id in ?", ids).Find(&sources).Error
	if err != nil {
		e.Log.Errorf("get SourceBasic error:%s \r\n", err)
		return sourceMap, err
	}
	for _, v := range sources {
		v := v
		sourceMap[v.Id] = v
	}
	return sourceMap, nil
}

func (e *SourceBasicService) DeleteCustomSource(basic live_schedule.SourceBasic) error {

	streamHubService := streamHubService.StreamHubService{}
	streamHubService.Log = e.Log

	err := streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: basic.TaskId})
	if err != nil {
		e.Log.Errorf("StopStreamHub error:%s \r\n", err)
		//tx.Rollback()
		return err
	}
	err = streamHubService.DeleteStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: basic.TaskId})
	if err != nil {
		e.Log.Errorf("StartStreamHub error:%s \r\n", err)
		//tx.Rollback()
		return err
	}

	err = e.Orm.Model(&live_schedule.SourceBasic{}).Where("id=?", basic.Id).Delete(&live_schedule.SourceBasic{}).Error
	if err != nil {
		e.Log.Errorf("delete source basic error:%s \r\n", err)
		//tx.Rollback()
		return err
	}

	err = e.Orm.Model(&live_schedule.SourceDetail{}).Where("source_id=?", basic.Id).Delete(&live_schedule.SourceDetail{}).Error
	if err != nil {
		e.Log.Errorf("delete source SourceDetail error:%s \r\n", err)
		//tx.Rollback()
		return err
	}
	return nil
}

func (e *SourceBasicService) AutoSwitchSource(c *dto.AutoSwitchReq) error {
	if len(c.SwitchId) <= 0 {
		return nil
	}
	param := make(map[string]interface{})
	active := c.ActiveIndex + 1
	param["active_item"] = active
	param["update_time"] = time.Now().Unix()

	//err := e.Orm.Model(&live_schedule.SourceBasic{}).Where("origin_switch_task_id=?", c.SwitchId).Updates(&param).Error
	//if err != nil {
	//	e.Log.Errorf("update SourceDetail error:%s", err)
	//	return err
	//}

	switchTask := &live_schedule.SwitchTask{}
	err := e.Orm.Model(&live_schedule.SwitchTask{}).Where("task_id=?", c.SwitchId).Find(&switchTask).Error
	if err != nil {
		e.Log.Errorf("get SwitchTask error:%s", err)
		return err
	}
	basic := &live_schedule.ChannelBasic{}
	err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", switchTask.ChannelId).Find(&basic).Error
	if err != nil {
		e.Log.Errorf("get ChannelBasic error:%s", err)
		return err
	}

	chartDat := live_schedule.ChartData{}
	if len(basic.ChartData) > 0 {
		err = json.Unmarshal([]byte(basic.ChartData), &chartDat)
		if err != nil {
			e.Log.Errorf("get ChartData error:%s", err)
			return err
		}
		for k, datum := range chartDat.NodeData {
			if datum.Node.Id == switchTask.MediaId {
				*(&chartDat.NodeData[k].Node.Data.ActiveIndex) = c.ActiveIndex
			}
		}
	}
	chartDatStr, err := json.Marshal(chartDat)
	if err != nil {
		e.Log.Errorf("Marshal ChartData error:%s", err)
		return err
	}

	err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Update("chart_data", chartDatStr).Error
	if err != nil {
		e.Log.Errorf("Marshal ChartData error:%s", err)
		return err
	}

	return nil
}

var m sync.Map

func (e *SourceBasicService) TimeStartSource() {

	// 使用 os.Stat 函数获取文件的信息
	_, err := os.Stat("/var/vs/failover/master")

	// 判断文件是否存在
	if os.IsNotExist(err) {
		e.Log.Errorf("文件不存在")
		return
	}

	basics := make([]live_schedule.SourceBasic, 0, 5)
	//获取所有信源
	err = e.Orm.Model(&live_schedule.SourceBasic{}).Where("is_delete=2").Find(&basics).Error
	if err != nil {
		e.Log.Errorf("get SourceBasic error:%s", err)
	}

	//信源状态
	sourceMap := make(map[int]live_schedule.SourceBasic)
	sourceIds := make([]int, 0, 5)
	for _, v := range basics {
		v := v
		sourceIds = append(sourceIds, v.Id)
		sourceMap[v.Id] = v
	}

	useMap := make(map[int]struct{})
	channelIds := make([]int, 0, 5)
	channelSourceMap := make(map[int][]int)
	channelIdMap := make(map[int]struct{})

	//获取当前所有任务
	convertTasks := make([]live_schedule.ChannelTask, 0, 5)
	err = e.Orm.Model(&live_schedule.ChannelTask{}).Where("source_id in ?", sourceIds).Find(&convertTasks).Error
	if err != nil {
		e.Log.Errorf("get ChannelTask error:%s", err)
	}
	for _, v := range convertTasks {
		v := v
		sources, ok := channelSourceMap[v.ChannelId]
		if !ok {
			sources = make([]int, 0, 5)
		}
		sources = append(sources, v.SourceId)
		channelSourceMap[v.ChannelId] = sources
		_, ok1 := channelIdMap[v.ChannelId]
		if !ok1 {
			channelIds = append(channelIds, v.ChannelId)
			channelIdMap[v.ChannelId] = struct{}{}
		}
		_, ok2 := useMap[v.SourceId]
		if !ok2 {
			useMap[v.SourceId] = struct{}{}
		}
	}

	//获取当前所有任务
	cortexTask := make([]live_schedule.CortexTask, 0, 5)
	err = e.Orm.Model(&live_schedule.CortexTask{}).Where("source_id in ?", sourceIds).Find(&cortexTask).Error
	if err != nil {
		e.Log.Errorf("get CortexTask error:%s", err)
	}
	for _, v := range cortexTask {
		sources, ok := channelSourceMap[v.ChannelId]
		if !ok {
			sources = make([]int, 0, 5)
		}
		sources = append(sources, v.SourceId)
		channelSourceMap[v.ChannelId] = sources
		_, ok1 := channelIdMap[v.ChannelId]
		if !ok1 {
			channelIds = append(channelIds, v.ChannelId)
			channelIdMap[v.ChannelId] = struct{}{}
		}
		_, ok2 := useMap[v.SourceId]
		if !ok2 {
			useMap[v.SourceId] = struct{}{}
		}

	}

	auditTasks := make([]live_schedule.AuditTask, 0, 5)
	err = e.Orm.Model(&live_schedule.AuditTask{}).Where("source_id in ?", sourceIds).Find(&auditTasks).Error
	if err != nil {
		e.Log.Errorf("get AuditTask error:%s", err)
	}
	for _, v := range auditTasks {
		sources, ok := channelSourceMap[v.ChannelId]
		if !ok {
			sources = make([]int, 0, 5)
		}
		sources = append(sources, v.SourceId)
		channelSourceMap[v.ChannelId] = sources
		_, ok1 := channelIdMap[v.ChannelId]
		if !ok1 {
			channelIds = append(channelIds, v.ChannelId)
			channelIdMap[v.ChannelId] = struct{}{}
		}
		_, ok2 := useMap[v.SourceId]
		if !ok2 {
			useMap[v.SourceId] = struct{}{}
		}
	}
	recordTasks := make([]live_schedule.SourceRecordTask, 0, 5)
	err = e.Orm.Model(&live_schedule.SourceRecordTask{}).Where("source_id in ? and record_type = 2", sourceIds).Find(&recordTasks).Error
	if err != nil {
		e.Log.Errorf("get SourceRecordTask error:%s", err)
	}
	for _, v := range recordTasks {
		sources, ok := channelSourceMap[v.ChannelId]
		if !ok {
			sources = make([]int, 0, 5)
		}
		sources = append(sources, v.SourceId)
		channelSourceMap[v.ChannelId] = sources
		_, ok1 := channelIdMap[v.ChannelId]
		if !ok1 {
			channelIds = append(channelIds, v.ChannelId)
			channelIdMap[v.ChannelId] = struct{}{}
		}
		_, ok2 := useMap[v.SourceId]
		if !ok2 {
			useMap[v.SourceId] = struct{}{}
		}
	}

	transCodeTasks := make([]live_schedule.ChannelTransCodeTask, 0, 5)
	err = e.Orm.Model(&live_schedule.ChannelTransCodeTask{}).Where("source_id in ?", sourceIds).Find(&transCodeTasks).Error
	if err != nil {
		e.Log.Errorf("get ChannelTransCodeTask error:%s", err)
	}
	for _, v := range transCodeTasks {
		sources, ok := channelSourceMap[v.ChannelId]
		if !ok {
			sources = make([]int, 0, 5)
		}
		sources = append(sources, v.SourceId)
		channelSourceMap[v.ChannelId] = sources
		_, ok1 := channelIdMap[v.ChannelId]
		if !ok1 {
			channelIds = append(channelIds, v.ChannelId)
			channelIdMap[v.ChannelId] = struct{}{}
		}
		_, ok2 := useMap[v.SourceId]
		if !ok2 {
			useMap[v.SourceId] = struct{}{}
		}

	}

	waterMarkTransTasks := make([]live_schedule.WaterMarkTransTask, 0, 5)
	err = e.Orm.Model(&live_schedule.WaterMarkTransTask{}).Where("source_id in ?", sourceIds).Find(&waterMarkTransTasks).Error
	if err != nil {
		e.Log.Errorf("get waterMarkTransTasks error:%s", err)
	}
	for _, v := range waterMarkTransTasks {
		sources, ok := channelSourceMap[v.ChannelId]
		if !ok {
			sources = make([]int, 0, 5)
		}
		sources = append(sources, v.SourceId)
		channelSourceMap[v.ChannelId] = sources
		_, ok1 := channelIdMap[v.ChannelId]
		if !ok1 {
			channelIds = append(channelIds, v.ChannelId)
			channelIdMap[v.ChannelId] = struct{}{}
		}
		_, ok2 := useMap[v.SourceId]
		if !ok2 {
			useMap[v.SourceId] = struct{}{}
		}
	}

	channelBasics := make([]live_schedule.ChannelBasic, 0, 5)
	err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id in ?", channelIds).Find(&channelBasics).Error
	if err != nil {
		e.Log.Errorf("get ChannelBasic error:%s", err)
	}
	sourceState := make(map[int]int)

	//查询调度任务中信源的状态
	for _, v := range channelBasics {
		v := v
		sources, _ := channelSourceMap[v.Id]
		for _, s := range sources {
			s := s
			state, _ := sourceState[s]
			//已经被调度使用的信源跳过判断
			if state == 1 {
				continue
			}
			if v.ChannelState != 2 {
				sourceState[s] = 1
			}

			if v.ChannelState == 2 {
				sourceState[s] = 2
			}
		}
	}

	monitorSourceMap := make(map[int][]int)
	monitorIdMap := make(map[int]struct{})
	monitorIds := make([]int, 0, 5)
	monitorConfigs := make([]monitor.MonitorConfig, 0, 5)
	err = e.Orm.Model(&monitor.MonitorConfig{}).Where("source_id in ?", sourceIds).Find(&monitorConfigs).Error
	if err != nil {
		e.Log.Errorf("get monitorConfigs error:%s", err)
	}
	for _, v := range monitorConfigs {
		sources, ok := monitorSourceMap[int(v.BasicId)]
		if !ok {
			sources = make([]int, 0, 5)
		}
		sources = append(sources, int(v.SourceId))
		monitorSourceMap[int(v.BasicId)] = sources
		_, ok1 := monitorIdMap[int(v.BasicId)]
		if !ok1 {
			monitorIds = append(monitorIds, int(v.BasicId))
			monitorIdMap[int(v.BasicId)] = struct{}{}
		}
		_, ok2 := useMap[int(v.SourceId)]
		if !ok2 {
			useMap[int(v.SourceId)] = struct{}{}
		}
	}

	monitorBasics := make([]monitor.MonitorBasic, 0, 5)
	err = e.Orm.Model(&monitor.MonitorBasic{}).Where("id in ?", monitorIds).Find(&monitorBasics).Error
	if err != nil {
		e.Log.Errorf("get MonitorBasic error:%s", err)
	}

	for _, v := range monitorBasics {
		v := v
		sources, _ := monitorSourceMap[v.Id]
		for _, s := range sources {
			s := s
			state, _ := sourceState[s]
			//已经被调度使用的信源跳过判断
			if state == 1 {
				continue
			}
			if v.MonitorState != 2 {
				sourceState[s] = 1
			}

			if v.MonitorState == 2 {
				sourceState[s] = 2
			}
		}
	}

	unusedSources := make([]live_schedule.SourceBasic, 0, 3)
	//查询信源在调度任务中的状态
	for _, b := range basics {
		b := b
		_, ok := useMap[b.Id]
		//已经被调度使用的信源跳过判断
		if ok {
			continue
		}
		if b.SourceState != 2 {
			unusedSources = append(unusedSources, b)
		}
	}
	for k, state := range sourceState {

		source, _ := sourceMap[k]

		if source.Kind == 4 {
			continue
		}
		if state == 1 && source.SourceState == 2 {
			e.StartSource(&dto.StartSourceReq{Id: source.Id, TenantId: source.TenantId})
		}

		if state == 2 && source.SourceState != 2 {
			e.StopSource(&dto.StopSourceReq{Id: source.Id, TenantId: source.TenantId})
		}
	}

	for _, v := range unusedSources {
		if v.Kind == 4 {
			continue
		}
		e.StopSource(&dto.StopSourceReq{Id: v.Id, TenantId: v.TenantId})
	}

}

func (e SourceBasicService) EquipBeat(d *dto.EquipBeatReq) error {
	now := time.Now().Unix()

	e.Log.Trace("EquipBeat回调：", d.EquipCode)
	err := e.Orm.Model(&live_schedule.SourceBasic{}).Where("equip_no=?", d.EquipCode).Update("equip_time", now).Error
	if err != nil {
		e.Log.Errorf("update SourceBasic error:%s \r\n", err)
	}
	return nil
}

func (e SourceBasicService) TimeStopEquip() {

	now := time.Now().Unix()

	offEquipNos := make([]string, 0, 3)
	onEquipNos := make([]string, 0, 3)
	stopEquipNos := make([]string, 0, 3)

	equips := make([]live_schedule.SourceBasic, 0, 5)
	err := e.Orm.Model(&live_schedule.SourceBasic{}).Where("equip_no !='' and equip_time>0").Find(&equips).Error
	if err != nil {
		e.Log.Errorf("update SourceBasic error:%s \r\n", err)
	}

	//此设备长时间没收到信号停止任务
	for _, equip := range equips {
		equip := equip
		if equip.EquipTime+30 < now {
			offEquipNos = append(offEquipNos, equip.EquipNo)
		} else {
			onEquipNos = append(onEquipNos, equip.EquipNo)
		}
		if equip.EquipTime+7200 < now {
			stopEquipNos = append(stopEquipNos, equip.EquipNo)
		}
	}

	if len(offEquipNos) > 0 {
		err := e.Orm.Model(&live_schedule.SourceBasic{}).Where("equip_no in ? and equip_state=1", offEquipNos).Update("equip_state", 2).Error
		if err != nil {
			e.Log.Errorf("update SourceBasic error:%s \r\n", err)
		}
	}

	if len(onEquipNos) > 0 {
		err := e.Orm.Model(&live_schedule.SourceBasic{}).Where("equip_no in ? and equip_state=2", onEquipNos).Update("equip_state", 1).Error
		if err != nil {
			e.Log.Errorf("update SourceBasic error:%s \r\n", err)
		}
	}

	if len(stopEquipNos) > 0 {
		basics := make([]live_schedule.SourceBasic, 0, 3)
		err := e.Orm.Model(&live_schedule.SourceBasic{}).Where("equip_no in ?", stopEquipNos).Find(&basics).Error
		if err != nil {
			e.Log.Errorf("get SourceBasic error:%s \r\n", err)
		}
		if len(basics) > 0 {
			for _, basic := range basics {
				if basic.SourceState != 2 {
					err = e.StopSource(&dto.StopSourceReq{Id: basic.Id})
					delete(models.EventMap, basic.EquipNo)
				}
			}

		}
	}

	return
}

func (e SourceBasicService) SaveSourceBasicNew(d *dto.SaveSourceBasicNewReq) error {
	sourceBasic := live_schedule.SourceBasicNew{}
	err := e.Orm.Model(&live_schedule.SourceBasicNew{}).Where("id=?", d.Id).Find(&sourceBasic).Error
	if err != nil {
		e.Log.Errorf("SaveSourceBasicNew error:%s \r\n", err)
		return err
	}
	d.Generate(&sourceBasic)
	err = e.Orm.Save(&sourceBasic).Error
	if err != nil {
		e.Log.Errorf("SaveSourceBasicNew error:%s \r\n", err)
		return err
	}
	return nil
}

func (e SourceBasicService) DelSourceBasicNew(d *dto.DelSourceBasicNewReq) error {
	sourceBasic := live_schedule.SourceBasicNew{}
	err := e.Orm.Model(&live_schedule.SourceBasicNew{}).Where("id=?", d.Id).Delete(&sourceBasic).Error
	if err != nil {
		e.Log.Errorf("DelSourceBasicNew error:%s \r\n", err)
		return err
	}
	return nil
}

func (e SourceBasicService) GetSourceBasicNew(d *dto.GetSourceBasicNewReq) (live_schedule.SourceBasicNew, error) {
	sourceBasic := live_schedule.SourceBasicNew{}
	err := e.Orm.Model(&live_schedule.SourceBasicNew{}).Where("id=?", d.Id).Find(&sourceBasic).Error
	if err != nil {
		e.Log.Errorf("GetSourceBasicNew error:%s \r\n", err)
		return sourceBasic, err
	}
	return sourceBasic, nil
}

func (e *SourceBasicService) GetSourceBasicNewPage(c *dto.SourceBasicNewGetPageReq, count *int64) (*[]live_schedule.SourceBasicNew, error) {
	var err error
	var data live_schedule.SourceBasicNew
	list := make([]live_schedule.SourceBasicNew, 0, 5)

	db := e.Orm.Model(&data).
		Scopes(
			cDto.MakeCondition(c.GetNeedSearch()),
			cDto.Paginate(c.GetPageSize(), c.GetPageIndex()),
		).Order("create_time desc")

	err = db.Find(&list).Limit(-1).Offset(-1).
		Count(count).Error
	if err != nil {
		e.Log.Errorf("SourceBasicService GetPage error:%s \r\n", err)
		return nil, err
	}
	return &list, nil
}
